[ Agent ] [ CMII ] - 新增大量功能 - 1

agent-go/k8s_exec/K8sOperator.go

@@ -13,6 +13,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"strings"
 	"sync"
+	"time"
 )
 
 var log = logger.Log
@@ -241,19 +242,6 @@ func (op *CmiiK8sOperator) DeploymentAllInterface(cmiiEnv string) []CmiiDeployme
 
 	index := 0
 	for deployment := range ccc {
-		// 过滤所有非CMII的内容
-		if strings.Contains(deployment.Name, "proxy") {
-			continue
-		}
-		if strings.Contains(deployment.Name, "minio") {
-			continue
-		}
-		if strings.HasPrefix(deployment.Name, "helm-live-rtsp") {
-			continue
-		}
-		if strings.Contains(deployment.Name, "nfs") {
-			continue
-		}
 		results[index] = deployment
 		index++
 	}
@@ -323,7 +311,7 @@ func (op *CmiiK8sOperator) DeploymentScale(cmiiEnv, appFizz string, scaleCount i
 				Namespace: deployment.Namespace,
 			},
 			Spec: autoscalingv1.ScaleSpec{
-				Replicas: int32(scaleCount),
+				Replicas: scaleCount,
 			},
 		}
 
@@ -379,6 +367,30 @@ func (op *CmiiK8sOperator) DeploymentUpdateTag(cmiiEnv, appName, newTag string)
 	return true
 }
 
+func (op *CmiiK8sOperator) DeploymentRestart(cmiiEnv, appName string) bool {
+	op.changeOperatorEnv(cmiiEnv)
+
+	result := true
+
+	deployment := op.DeploymentOneInterface(cmiiEnv, appName)
+	if deployment == nil {
+		log.ErrorF("[DeploymentRestart] - [%s] [%s] not exists !", cmiiEnv, appName)
+		return false
+	}
+
+	result = op.DeploymentScale(deployment.Namespace, deployment.Name, 0)
+	if !result {
+		return result
+	}
+	time.Sleep(time.Millisecond * 200)
+	result = op.DeploymentScale(deployment.Namespace, deployment.Name, deployment.Replicas)
+	if !result {
+		return result
+	}
+
+	return result
+}
+
 func (op *CmiiK8sOperator) ReplicaSet(cmiiEnv, replicaSetName string) *v1.ReplicaSet {
 
 	op.changeOperatorEnv(cmiiEnv)
@@ -451,7 +463,69 @@ func (op *CmiiK8sOperator) PodAll(cmiiEnv string) []corev1.Pod {
 		index++
 	}
 
-	return results
+	return results[:index]
+}
+
+func (op *CmiiK8sOperator) PodAllInterface(cmiiEnv string) []CmiiPodInterface {
+	op.changeOperatorEnv(cmiiEnv)
+	client := op.CurrentClient
+
+	podList, err := client.CoreV1().Pods(op.CurrentNamespace).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		log.ErrorF("[PodAll] - list c in [%s] [%s] error => %s", cmiiEnv, op.CurrentNamespace, err.Error())
+		return nil
+	}
+	pods := podList.Items
+
+	length := len(pods)
+	results := make([]CmiiPodInterface, length)
+	ccc := make(chan CmiiPodInterface, length)
+	var wg sync.WaitGroup
+	podInterface := CmiiPodInterface{}
+
+	worker := workerThread
+	if length <= worker {
+		for i, pod := range pods {
+			objectMeta := pod.ObjectMeta
+			objectMeta.SetAnnotations(nil)
+			objectMeta.SetManagedFields(nil)
+			pod.ObjectMeta = objectMeta
+
+			results[i] = podInterface.Convert(pod)
+		}
+		return results
+	}
+
+	pinch := length / worker
+	wg.Add(worker)
+
+	for splice := 0; splice < worker; splice++ {
+		go func(podList []corev1.Pod) {
+			defer wg.Done()
+			for _, pod := range podList {
+				objectMeta := pod.ObjectMeta
+				objectMeta.SetAnnotations(nil)
+				objectMeta.SetManagedFields(nil)
+				pod.ObjectMeta = objectMeta
+
+				ccc <- podInterface.Convert(pod)
+			}
+
+		}(pods[splice*pinch : utils.MinInt((splice+1)*pinch, length)])
+	}
+
+	go func() {
+		wg.Wait()
+		close(ccc)
+	}()
+
+	index := 0
+	for c := range ccc {
+		results[index] = c
+		index++
+	}
+
+	return results[:index]
 }
 
 func (op *CmiiK8sOperator) PodFizz(cmiiEnv, appFizz string) (fizzPod []corev1.Pod) {