Merge pull request #20 from caoxianfei1/master

figure out etcd peerPort and clientPort confused definition and revise

api/v1/curvecluster_types.go

@@ -207,10 +207,10 @@ type CurveVersionSpec struct {
 // EtcdSpec
 type EtcdSpec struct {
 	// +optional
-	Port int `json:"port,omitempty"`
+	PeerPort int `json:"peerPort,omitempty"`
 
 	// +optional
-	ListenPort int `json:"listenPort,omitempty"`
+	ClientPort int `json:"clientPort,omitempty"`
 
 	// +optional
 	Config map[string]string `json:"config,omitempty"`

config/crd/bases/operator.curve.io_curveclusters.yaml

@@ -75,13 +75,13 @@ spec:
             etcd:
               description: EtcdSpec
               properties:
+                clientPort:
+                  type: integer
                 config:
                   additionalProperties:
                     type: string
                   type: object
-                listenPort:
-                  type: integer
-                port:
+                peerPort:
                   type: integer
               type: object
             logDirHostPath:

config/deploy/manifests.yaml

@@ -80,13 +80,13 @@ spec:
             etcd:
               description: EtcdSpec
               properties:
+                clientPort:
+                  type: integer
                 config:
                   additionalProperties:
                     type: string
                   type: object
-                listenPort:
-                  type: integer
-                port:
+                peerPort:
                   type: integer
               type: object
             logDirHostPath:
@@ -423,7 +423,7 @@ spec:
         - --enable-leader-election
         command:
         - ./curve-operator
-        image: harbor.cloud.netease.com/curve/curve-operator:8f3e0bd
+        image: harbor.cloud.netease.com/curve/curve-operator:684ba2b
         name: curve-operator
         resources:
           limits:

config/manager/kustomization.yaml

@@ -5,4 +5,4 @@ kind: Kustomization
 images:
 - name: harbor.cloud.netease.com/curve/curve-operator
   newName: harbor.cloud.netease.com/curve/curve-operator
-  newTag: 8f3e0bd
+  newTag: 684ba2b

config/samples/operator_v1_curvecluster.yaml → config/samples/cluster.yaml

@@ -4,33 +4,32 @@ metadata:
   name: my-cluster
   namespace: curvebs
 spec:
-  # Add fields here
   curveVersion:
     image: opencurvedocker/curvebs:v1.2
   nodes:
-  - node1
-  - node2
-  - node3
-  dataDirHostPath: "/curvebs/data"
-  logDirHostPath: "/curvebs/log"
+  - curve-operator-node1
+  - curve-operator-node2
+  - curve-operator-node3
+  dataDirHostPath: /curvebs/data
+  logDirHostPath: /curvebs/log
   etcd:
-    port: 23880
-    listenPort: 23890
+    peerPort: 23891
+    clientPort: 23791
   mds:
     port: 23970
     dummyPort: 23960
   storage:
     useSelectedNodes: false
     nodes:
-    - node1
-    - node2
-    - node3
+    - curve-operator-node1
+    - curve-operator-node2
+    - curve-operator-node3
     port: 8200
     copysets: 100
     devices:
-    - name: "/dev/sdb"
-      mountPath: "/data/chunkserver0"
-      percentage: 90
+    - name: /dev/vdc
+      mountPath: /data/chunkserver0
+      percentage: 80
   snapShotClone:
     enable: true
     port: 5555
@@ -39,6 +38,6 @@ spec:
     s3Config:
       ak: minioadmin
       sk: minioadmin
-      nosAddress: http://10.219.196.145:9000
+      nosAddress: http://59.111.93.76:9001
       bucketName: curvebs
 

delete-resource.sh

@@ -58,13 +58,11 @@ kubectl delete deploy curve-snapshotclone-c -n curvebs
 
 # all job in curvebs cluster
 kubectl delete --all job -n curvebs
+# all deploy in curvebs cluster
+kubectl delete --all deployment -n curvebs
 
 # all po in curvebs cluster
 kubectl delete --all pods -n curvebs
 
-
-# all deploy in curvebs cluster
-kubectl delete --all deployment -n curvebs
-
 # curvecluster cr
 kubectl delete -f config/samples/

pkg/chunkserver/chunkserver.go

@@ -72,33 +72,34 @@ func (c *Cluster) Start(nodeNameIP map[string]string) error {
 
 	// 2. wait all job finish to complete format and wait MDS election success.
 	k8sutil.UpdateCondition(context.TODO(), &c.context, c.namespacedName, curvev1.ConditionTypeFormatedReady, curvev1.ConditionTrue, curvev1.ConditionFormatingChunkfilePoolReason, "Formating chunkfilepool")
-	halfMinuteTicker := time.NewTicker(10 * time.Second)
+	halfMinuteTicker := time.NewTicker(1 * time.Minute)
 	defer halfMinuteTicker.Stop()
 
 	chn := make(chan bool, 1)
-	ctx, canf := context.WithTimeout(context.Background(), time.Duration(10*60*time.Second))
+	ctx, canf := context.WithTimeout(context.Background(), time.Duration(24*60*60*time.Second))
 	defer canf()
 	c.checkJobStatus(ctx, halfMinuteTicker, chn)
 
-	// block here unitl timeout(10 mins) or all jobs has been successed.
+	// block here unitl timeout(24 hours) or all jobs has been successed.
 	flag := <-chn
 
 	// not all job has completed
 	if !flag {
 		// TODO: delete all jobs that has created.
-		log.Error("Format job is not completed in 10 minutes and exit with -1")
-		return errors.New("Format job is not completed in 10 minutes and exit with -1")
+		log.Error("Format job is not completed in 24 hours and exit with -1")
+		return errors.New("Format job is not completed in 24 hours and exit with -1")
 	}
 	k8sutil.UpdateCondition(context.TODO(), &c.context, c.namespacedName, curvev1.ConditionTypeFormatedReady, curvev1.ConditionTrue, curvev1.ConditionFormatChunkfilePoolReason, "Formating chunkfilepool successed")
 
-	log.Info("all jobs run completed in 10 mins")
+	log.Info("all jobs run completed in 24 hours")
 
 	// 2. create physical pool
 	_, err = c.runCreatePoolJob(nodeNameIP, "physical_pool")
 	if err != nil {
 		log.Error("failed to create physical pool")
 		return errors.Wrap(err, "failed to create physical pool")
 	}
+	log.Info("create physical pool successed")
 
 	// 3. startChunkServers start all chunkservers for each device of every node
 	err = c.startChunkServers()
@@ -108,6 +109,7 @@ func (c *Cluster) Start(nodeNameIP map[string]string) error {
 	}
 
 	// 4. wait all chunkservers online before create logical pool
+	log.Info("starting all chunkserver")
 	time.Sleep(30 * time.Second)
 
 	// 5. create logical pool
@@ -116,6 +118,7 @@ func (c *Cluster) Start(nodeNameIP map[string]string) error {
 		log.Error("failed to create logical pool")
 		return errors.Wrap(err, "failed to create physical pool")
 	}
+	log.Info("create logical pool successed")
 
 	k8sutil.UpdateCondition(context.TODO(), &c.context, c.namespacedName, curvev1.ConditionTypeChunkServerReady, curvev1.ConditionTrue, curvev1.ConditionChunkServerClusterCreatedReason, "Chunkserver cluster has been created")
 
@@ -127,7 +130,7 @@ func (c *Cluster) checkJobStatus(ctx context.Context, ticker *time.Ticker, chn c
 	for {
 		select {
 		case <-ticker.C:
-			log.Info("time is up")
+			log.Info("time is up(1 minute)")
 			completed := 0
 			for _, jobName := range jobsArr {
 				job, err := c.context.Clientset.BatchV1().Jobs(c.namespacedName.Namespace).Get(jobName, metav1.GetOptions{})

pkg/chunkserver/spec.go

@@ -128,17 +128,18 @@ func (c *Cluster) CreateS3ConfigMap() error {
 		return errors.Wrapf(err, "failed to get configmap %s from cluster", config.S3ConfigMapTemp)
 	}
 
+	data := s3CMTemplate.Data
 	// if true
 	if c.spec.SnapShotClone.Enable {
-		s3CMTemplate.Data["s3.ak"] = c.spec.SnapShotClone.S3Config.AK
-		s3CMTemplate.Data["s3.sk"] = c.spec.SnapShotClone.S3Config.SK
-		s3CMTemplate.Data["s3.nos_address"] = c.spec.SnapShotClone.S3Config.NosAddress
-		s3CMTemplate.Data["s3.snapshot_bucket_name"] = c.spec.SnapShotClone.S3Config.SnapShotBucketName
+		data["s3.ak"] = c.spec.SnapShotClone.S3Config.AK
+		data["s3.sk"] = c.spec.SnapShotClone.S3Config.SK
+		data["s3.nos_address"] = c.spec.SnapShotClone.S3Config.NosAddress
+		data["s3.snapshot_bucket_name"] = c.spec.SnapShotClone.S3Config.SnapShotBucketName
 	}
 
 	var configMapData string
-	for k, v := range s3CMTemplate.Data {
-		configMapData += configMapData + k + "=" + v + "\n"
+	for k, v := range data {
+		configMapData = configMapData + k + "=" + v + "\n"
 	}
 
 	s3ConfigMap := map[string]string{

pkg/controllers/cluster.go

@@ -66,7 +66,7 @@ func (c *cluster) reconcileCurveDaemons() error {
 
 	logger.Info("starting read config template job")
 
-	ticker := time.NewTicker(5 * time.Second)
+	ticker := time.NewTicker(10 * time.Second)
 	defer ticker.Stop()
 
 	chn := make(chan bool, 1)

pkg/etcd/etcd.go

@@ -50,14 +50,13 @@ func New(context clusterd.Context, namespacedName types.NamespacedName, spec cur
 func (c *Cluster) Start(nodeNameIP map[string]string) error {
 	var etcdEndpoints string
 	var clusterEtcdAddr string
-	var initial_cluster string
+
 	for _, ipAddr := range nodeNameIP {
-		etcdEndpoints = fmt.Sprint(etcdEndpoints, ipAddr, ":", c.spec.Etcd.Port, ",")
-		clusterEtcdAddr = fmt.Sprint(clusterEtcdAddr, ipAddr, ":", c.spec.Etcd.ListenPort, ",")
+		etcdEndpoints = fmt.Sprint(etcdEndpoints, ipAddr, ":", c.spec.Etcd.PeerPort, ",")
+		clusterEtcdAddr = fmt.Sprint(clusterEtcdAddr, ipAddr, ":", c.spec.Etcd.ClientPort, ",")
 	}
 	etcdEndpoints = strings.TrimRight(etcdEndpoints, ",")
 	clusterEtcdAddr = strings.TrimRight(clusterEtcdAddr, ",")
-	initial_cluster = strings.TrimRight(initial_cluster, ",")
 
 	// Create etcd override configmap
 	err := c.createOverrideConfigMap(etcdEndpoints, clusterEtcdAddr)
@@ -87,8 +86,9 @@ func (c *Cluster) Start(nodeNameIP map[string]string) error {
 	}
 
 	hostId := 0
+	var initial_cluster string
 	for _, nodeName := range nodeNamesOrdered {
-		initial_cluster = fmt.Sprint(initial_cluster, "etcd", strconv.Itoa(hostId), "0", "=http://", nodeNameIP[nodeName], ":", c.spec.Etcd.Port, ",")
+		initial_cluster = fmt.Sprint(initial_cluster, "etcd", strconv.Itoa(hostId), "0", "=http://", nodeNameIP[nodeName], ":", c.spec.Etcd.PeerPort, ",")
 		hostId++
 	}
 	initial_cluster = strings.TrimRight(initial_cluster, ",")
@@ -107,8 +107,8 @@ func (c *Cluster) Start(nodeNameIP map[string]string) error {
 			ServiceHostSequence:    strconv.Itoa(daemonID),
 			ServiceReplicaSequence: strconv.Itoa(replicasSequence),
 			ServiceAddr:            nodeNameIP[nodeName],
-			ServicePort:            strconv.Itoa(c.spec.Etcd.Port),
-			ServiceClientPort:      strconv.Itoa(c.spec.Etcd.ListenPort),
+			ServicePort:            strconv.Itoa(c.spec.Etcd.PeerPort),
+			ServiceClientPort:      strconv.Itoa(c.spec.Etcd.ClientPort),
 			ClusterEtcdHttpAddr:    initial_cluster,
 
 			DaemonID:             daemonIDString,