Skywalking Swck Agent注入實現分析

專案地址：

GitHub - apache/skywalking-swck: Apache SkyWalking Cloud on Kubernetes

專案簡介：

A bridge project between Apache SkyWalking and Kubernetes.
SWCK is a platform for the SkyWalking user that provisions, upgrades, maintains SkyWalking relevant components, and makes them work natively on Kubernetes.
skywalking-swck是一個在skywalking和kubernetes之間架起一座橋樑性質的專案。可以給使用者提供skywalking相關元件及後期升級、維護。讓他們使用起來更加雲原生。

專案特性

Java Agent Injector: Inject the java agent into the application pod natively.
- Inject the java agent into the application pod.
- Leverage a global configuration to simplify the agent and injector setup.
- Use the annotation to customize specific workloads.
- Synchronize injecting status to JavaAgent CR for monitoring purposes.
Operator: Provision and maintain SkyWalking backend components.
Custom Metrics Adapter: Provides custom metrics coming from SkyWalking OAP cluster for autoscaling by Kubernetes HPA
注入Java Agent：以更加雲原生的試注入java agent
- 將java agent注入到應用pod
- 可以全域性化設定，簡化agent注入操作
- 使用annotation自定義一些設定
- 將注入的狀態同步到JavaAgent這個CR物件，便於監控
Operator: 提供和維護SkyWalking後端的元件
自定義指標適配：能夠提供來自於SkyWalking OAP的自定義指標給kubernetes HPA,以便自動擴縮容

使用

skywalking-swck/java-agent-injector.md at master · apache/skywalking-swck

下載並安裝Operator
建立ConfigMap/SwAgent等全域性設定
Annotations > SwAgent > Configmap (Deprecated) > Default Configmap (Deprecated)
接入應用設定label/annotations
skywalking agent inject on kubernetes

原理

概述

當kubectl apply 一個 deployment資源後,k8s會建立pod，此時k8s根據mutatingwebhookconfigurations資源設定(設定了監控的資源以及webhook server資訊)，呼叫相應的webhook server，webhook server會進行處理，在pod yaml中注入initContainer設定，使業務容器與initContainer容器共用skywalking agent目錄，並且設定JAVA_TOOL_OPTIONS環境變數值為"-javaagent:/sky/agent/skywalking-agent.jar=agent.service_name=xxxx",這樣JVM啟動時，會附加上javaagent,以達到目的。

詳述

首先我們來看一下MutatingWebhookConfiguration和 ValidatingWebhookConfiguration資源
檢視kubectl explain 對這兩個資源的描述

MutatingWebhookConfiguration describes the configuration of and admission
     webhook that accept or reject and may change the object.
     
ValidatingWebhookConfiguration describes the configuration of and admission
     webhook that accept or reject and object without changing it.

簡而言之，這兩種資源都是准入控制器(Admission Controller)的兩種實現，都能控制是否接受還是拒絕對資源物件的變化，但不同的是，MutatingWebhookConfiguration可以改變資源物件，而ValidatingWebhookConfiguration不可以，可以參看搞懂 Kubernetes 准入控制（Admission Controller)詳細瞭解。
2. swck就是利用MutatingWebhookConfiguration實現了對pod的修改，我們來看下swck中的定義

apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
  annotations:
    cert-manager.io/inject-ca-from: skywalking-swck-system/skywalking-swck-serving-cert
  name: skywalking-swck-mutating-webhook-configuration
webhooks:
- admissionReviewVersions:
  - v1
  clientConfig:
    service:
      name: skywalking-swck-webhook-service
      namespace: skywalking-swck-system
      path: /mutate-v1-pod
  failurePolicy: Fail
  name: mpod.kb.io
  namespaceSelector:
    matchLabels:
      swck-injection: enabled
  rules:
  - apiGroups:
    - ""
    apiVersions:
    - v1
    operations:
    - CREATE
    - UPDATE
    resources:
    - pods
  sideEffects: None

從這段定義中可以看出，當帶有標籤swck-injection: enabled的Namespace下的POD資源有CREATE或者UPDATE操作時，將會呼叫path: /mutate-v1-pod。
3. 在swck專案中的operator/man.go中找到此URL

// register a webhook to enable the java agent injector
setupLog.Info("registering /mutate-v1-pod webhook")
mgr.GetWebhookServer().Register("/mutate-v1-pod",
   &webhook.Admission{
      Handler: &injector.JavaagentInjector{Client: mgr.GetClient()}})
setupLog.Info("/mutate-v1-pod webhook is registered")

swck向k8s註冊了/mutate-v1-pod以及對應的handler。我們可以想到，當create或update pod時，k8s將會呼叫path對應的handler處理。
4. 檢視Handler: injector.JavaagentInjector

// Handle will process every coming pod under the
// specified namespace which labeled "swck-injection=enabled"
func (r *JavaagentInjector) Handle(ctx context.Context, req admission.Request) admission.Response {
   pod := &corev1.Pod{}
   if err := r.decoder.Decode(req, pod); err != nil {
      return admission.Errored(http.StatusBadRequest, err)
   }

   // set Annotations to avoid repeated judgments
   if pod.Annotations == nil {
      pod.Annotations = map[string]string{}
   }

   // 查詢所有匹配的swAgent
   swAgentL := r.findMatchedSwAgentL(ctx, req, pod)

   //初始化所有annotations(載入所有annotations)
   anno, err := NewAnnotations()
   if err != nil {
      javaagentInjectorLog.Error(err, "get NewAnnotations error")
   }
   //建立AnnotationOverlay物件,它是一個map，用於儲存被overlaied的annotation
   ao := NewAnnotationOverlay()
   //建立SidecarInjectField物件
   s := NewSidecarInjectField()
   //構建inject鏈物件
   ip := NewInjectProcess(ctx, s, anno, ao, swAgentL, pod, req, javaagentInjectorLog, r.Client)
   //開始inject
   return ip.Run()
}

建立inject chain物件
Inject chain物件集合了此次有變更的Pod, webhook request, k8s client以及註解、swagent等物件

// NewInjectProcess create a new InjectProcess
func NewInjectProcess(ctx context.Context, injectFileds *SidecarInjectField, annotation *Annotations,
   annotationOverlay *AnnotationOverlay, swAgentL *v1alpha1.SwAgentList, pod *corev1.Pod, req admission.Request, log logr.Logger,
   kubeclient client.Client) *InjectProcessData {
   return &InjectProcessData{
      ctx:               ctx,
      injectFileds:      injectFileds,
      annotation:        annotation,
      annotationOverlay: annotationOverlay,
      swAgentL:          swAgentL,
      pod:               pod,
      req:               req,
      log:               log,
      kubeclient:        kubeclient,
   }
}

看下ip.Run()方法
經過了前面鋪墊，終於到了主題了，run方法首先按照執行順序倒序構造了一個執行鏈，然後執行。

// Run will connect the above six steps into a chain and start to execute the first step
func (ipd *InjectProcessData) Run() admission.Response {
   // set final step
   podInject := &PodInject{}

   // set next step is PodInject
   getConfigmap := &GetConfigmap{}
   getConfigmap.setNext(podInject)

   // set next step is GetConfigmap
   overlayPlugins := &OverlayPlugins{}
   overlayPlugins.setNext(getConfigmap)

   // set next step is OverlayPlugins
   overlayAgent := &OverlayAgent{}
   overlayAgent.setNext(overlayPlugins)

   // set next step is OverlayAgent
   overlaysidecar := &OverlaySidecar{}
   overlaysidecar.setNext(overlayAgent)

   // set next step is OverlaySwAgentCR
   overlaySwAgentCR := &OverlaySwAgentCR{}
   overlaySwAgentCR.setNext(overlaysidecar)

   // set next step is OverlaySidecar
   getStrategy := &GetStrategy{}
   getStrategy.setNext(overlaySwAgentCR)

   // this is first step and do real injection
   return getStrategy.execute(ipd)
}

首先執行的是GetStrategy

func (gs *GetStrategy) execute(ipd *InjectProcessData) admission.Response {
   log.Info("=============== GetStrategy ================")
   ipd.injectFileds.GetInjectStrategy(*ipd.annotation, &ipd.pod.ObjectMeta.Labels, &ipd.pod.ObjectMeta.Annotations)
   if !ipd.injectFileds.NeedInject {
      log.Info("don't inject agent")
      return admission.Allowed("ok")
   }
   return gs.next.execute(ipd)
}

// GetInjectStrategy gets user's injection strategy
func (s *SidecarInjectField) GetInjectStrategy(a Annotations, labels,
   annotation *map[string]string) {
   // set default value
   s.NeedInject = false

   // set NeedInject's value , if the pod has the label "swck-java-agent-injected=true", means need inject
   if *labels == nil {
      return
   }

   if strings.EqualFold((*labels)[ActiveInjectorLabel], "true") {
      s.NeedInject = true
   }

   if *annotation == nil {
      return
   }

   // set injectContainer's value
   if v, ok := (*annotation)[sidecarInjectContainerAnno]; ok {
      s.InjectContainer = v
   }
}

邏輯比較簡單，判斷當前pod需不需要inject,以及獲取inject的container名字的正規表示式。
8. 第二個執行的是OverlaySwAgentCR

// get configs from SwAgent CR
func (gs *OverlaySwAgentCR) execute(ipd *InjectProcessData) admission.Response {
   log.Info(fmt.Sprintf("=============== OverlaySwAgentCR(%d) ================ ", len(ipd.swAgentL.Items)))
   if !ipd.injectFileds.OverlaySwAgentCR(ipd.swAgentL, ipd.pod) {
      log.Info("overlay SwAgent cr config error.")
      return PatchReq(ipd.pod, ipd.req)
   }
   return gs.next.execute(ipd)
}

func (s *SidecarInjectField) OverlaySwAgentCR(swAgentL *v1alpha1.SwAgentList, pod *corev1.Pod) bool {
   s.ConfigmapVolume.ConfigMap = new(corev1.ConfigMapVolumeSource)

   // 如果找到多個匹配的SwAgent,則應用最後一個，其它幾個忽略
   if len(swAgentL.Items) > 0 {
      swAgent := swAgentL.Items[len(swAgentL.Items)-1]
      log.Info(fmt.Sprintf("agent %s loaded.", swAgent.Name))
      // 首先設定了shared volume, mount path。預設sharedVolumeName是sky-agent,mount path是/sky/agent
      // volume name可以更改, mountPath無法更改, mount path是業務容器上的path
      s.SidecarVolume.Name = swAgent.Spec.SharedVolumeName
      s.SidecarVolume.VolumeSource.EmptyDir = &corev1.EmptyDirVolumeSource{}
      s.SidecarVolumeMount.Name = swAgent.Spec.SharedVolumeName
      s.SidecarVolumeMount.MountPath = mountPath

      // 如果swagent設定了configmap,則設定業務容器mount path,實際是由configmap中的agent設定
      // 覆蓋agent映象中的設定
      if swAgent.Spec.SwConfigMapVolume != nil {
         if len(swAgent.Spec.SwConfigMapVolume.Name) > 0 &&
            len(swAgent.Spec.SwConfigMapVolume.ConfigMapName) > 0 &&
            len(swAgent.Spec.SwConfigMapVolume.ConfigMapMountFile) > 0 {
            //s.ConfigmapVolume = corev1.Volume{}
            s.ConfigmapVolume.Name = swAgent.Spec.SwConfigMapVolume.Name
            s.ConfigmapVolume.ConfigMap = new(corev1.ConfigMapVolumeSource)
            s.ConfigmapVolume.ConfigMap.Name = swAgent.Spec.SwConfigMapVolume.ConfigMapName
            //s.ConfigmapVolumeMount = corev1.VolumeMount{}
            s.ConfigmapVolumeMount.Name = swAgent.Spec.SwConfigMapVolume.Name
            s.ConfigmapVolumeMount.MountPath = "/sky/agent/config/" + swAgent.Spec.SwConfigMapVolume.ConfigMapMountFile
            s.ConfigmapVolumeMount.SubPath = swAgent.Spec.SwConfigMapVolume.ConfigMapMountFile
         }
      }

      // init container
      s.Initcontainer.Name = swAgent.Spec.JavaSidecar.Name
      s.Initcontainer.Image = swAgent.Spec.JavaSidecar.Image
      s.Initcontainer.Args = swAgent.Spec.JavaSidecar.Args
      s.Initcontainer.Command = swAgent.Spec.JavaSidecar.Command
      s.Initcontainer.VolumeMounts = append(s.Initcontainer.VolumeMounts, corev1.VolumeMount{
         Name:      swAgent.Spec.SharedVolumeName,
         MountPath: mountPath,
      })

      // 將swagent設定的環境變數設定為業務容器的環境變數
      s.Envs = swAgent.Spec.JavaSidecar.Env
      s.InjectContainer = swAgent.Spec.ContainerMatcher
   }

   return true
}

第三個執行的是OverlaySidecar

func (os *OverlaySidecar) execute(ipd *InjectProcessData) admission.Response {
   log.Info("=============== OverlaySidecar ================")
   if !ipd.injectFileds.OverlaySidecar(*ipd.annotation, ipd.annotationOverlay, &ipd.pod.ObjectMeta.Annotations) {
      return PatchReq(ipd.pod, ipd.req)
   }
   return os.next.execute(ipd)
}

// OverlaySidecar overlays default config
func (s *SidecarInjectField) OverlaySidecar(a Annotations, ao *AnnotationOverlay, annotation *map[string]string) bool {
   s.Initcontainer.Command = make([]string, 1)
   s.Initcontainer.Args = make([]string, 2)
   if nil == s.ConfigmapVolume.ConfigMap {
      s.ConfigmapVolume.ConfigMap = new(corev1.ConfigMapVolumeSource)
   }

   limitsStr := ""
   requestStr := ""
   // 建立sidercar註解map物件,其初始值從上一步執行結果中獲取.map中的key為sidecar註解去掉字首後的名稱
   annoField := map[string]*string{
      "initcontainer.Name":               &s.Initcontainer.Name,
      "initcontainer.Image":              &s.Initcontainer.Image,
      "initcontainer.Command":            &s.Initcontainer.Command[0],
      "initcontainer.args.Option":        &s.Initcontainer.Args[0],
      "initcontainer.args.Command":       &s.Initcontainer.Args[1],
      "initcontainer.resources.limits":   &limitsStr,
      "initcontainer.resources.requests": &requestStr,
      "sidecarVolume.Name":               &s.SidecarVolume.Name,
      "sidecarVolumeMount.MountPath":     &s.SidecarVolumeMount.MountPath,
      "configmapVolume.ConfigMap.Name":   &s.ConfigmapVolume.ConfigMap.Name,
      "configmapVolume.Name":             &s.ConfigmapVolume.Name,
      "configmapVolumeMount.MountPath":   &s.ConfigmapVolumeMount.MountPath,
      "env.Name":                         &s.Env.Name,
      "env.Value":                        &s.Env.Value,
   }

    // 從全量註解中獲取sidercar字首的註解，遍歷，檢查Pod有沒有設定相應sidercar註解，如果設定了，則覆蓋map中對應key原來的值
   anno := GetAnnotationsByPrefix(a, sidecarAnnotationPrefix)
   for _, v := range anno.Annotations {
      fieldName := strings.TrimPrefix(v.Name, sidecarAnnotationPrefix)
      if pointer, ok := annoField[fieldName]; ok {
         if !s.setValue(pointer, ao, annotation, v) {
            return false
         }
      }
   }

   s.SidecarVolumeMount.Name = s.SidecarVolume.Name
   s.ConfigmapVolumeMount.Name = s.ConfigmapVolume.Name
   s.Initcontainer.VolumeMounts = []corev1.VolumeMount{s.SidecarVolumeMount}

   // 設定init container的資源限制
   if limitsStr != "nil" {
      limits := make(corev1.ResourceList)
      err := json.Unmarshal([]byte(limitsStr), &limits)
      if err != nil {
         log.Error(err, "unmarshal limitsStr error")
         return false
      }
      s.Initcontainer.Resources.Limits = limits
   }

   // 設定init container需要申請的資源
   if requestStr != "nil" {
      requests := make(corev1.ResourceList)
      err := json.Unmarshal([]byte(requestStr), &requests)
      if err != nil {
         log.Error(err, "unmarshal requestStr error")
         return false
      }
      s.Initcontainer.Resources.Requests = requests
   }

   // the sidecar volume's type is determined
   s.SidecarVolume.VolumeSource.EmptyDir = nil

   return true
}

第四個執行的是OverlayAgent

// OverlayAgent overlays the agent by getting the pod's annotations
// If the agent overlay option is not set, go directly to the next step
// If set the wrong value in the annotation , inject the error info and return
func (oa *OverlayAgent) execute(ipd *InjectProcessData) admission.Response {
   log.Info("=============== OverlayAgent ================")
   if !ipd.injectFileds.OverlayAgent(*ipd.annotation, ipd.annotationOverlay, &ipd.pod.ObjectMeta.Annotations) {
      ipd.log.Info("overlay agent config error!please look the error annotation!")
      return PatchReq(ipd.pod, ipd.req)
   }
   return oa.next.execute(ipd)
}

// OverlayAgent overlays agent
func (s *SidecarInjectField) OverlayAgent(a Annotations, ao *AnnotationOverlay, annotation *map[string]string) bool {
   // jvmAgentConfigStr init
   s.JvmAgentConfigStr = ""
   
   //遍歷pod的註解，如果註解的名稱存在於全量註解中，則將Pod註解及值儲存到AnnotationOverlay map物件中
   anno := GetAnnotationsByPrefix(a, agentAnnotationPrefix)
   for k, v := range *annotation {
      if strings.HasPrefix(k, agentAnnotationPrefix) {
         for _, an := range anno.Annotations {
            if strings.EqualFold(k, an.Name) {
               if !s.AgentOverlayandGetValue(ao, annotation, an) {
                  return false
               }
            }
         }
         
         // 將pod註解去掉agent字首,追加到JvmAgentConfigStr欄位中
         configName := strings.TrimPrefix(k, agentAnnotationPrefix)
         config := strings.Join([]string{configName, v}, "=")
         // add to jvmAgentConfigStr
         if s.JvmAgentConfigStr != "" {
            s.JvmAgentConfigStr = strings.Join([]string{s.JvmAgentConfigStr, config}, ",")
         } else {
            s.JvmAgentConfigStr = config
         }
      }
   }
   return true
}

第五個執行的是OverlayPlugins,與OverlayAgent邏輯類似。
第六個執行的是GetConfigmap,其作用是檢查如果pod設定了agent configmap,則檢查configmap設定的值是否正確.

func (s *SidecarInjectField) ValidateConfigmap(ctx context.Context, kubeclient client.Client, namespace string,
   annotation *map[string]string) bool {
   if len(s.ConfigmapVolume.Name) == 0 || len(s.ConfigmapVolume.ConfigMap.Name) == 0 {
      return true
   }
   configmap := &corev1.ConfigMap{}
   configmapName := s.ConfigmapVolume.VolumeSource.ConfigMap.LocalObjectReference.Name
   // check whether the configmap is existed
   err := kubeclient.Get(ctx, client.ObjectKey{Namespace: namespace, Name: configmapName}, configmap)
   if err != nil && !errors.IsNotFound(err) {
      log.Error(err, "Get Configmap failed", "configmapName", configmapName, "namespace", namespace)
      return false
   }
   // if configmap exist , validate it
   if !errors.IsNotFound(err) {
      ok, errinfo := ValidateConfigmap(configmap)
      if ok {
         log.Info("the configmap validate true", "configmapName", configmapName)
         return true
      }
      log.Error(errinfo, "the configmap validate false", "configmapName", configmapName)
   }
   return true
}

最後一步是PodInject,顧名思義，其作用是進行Pod注入

// PodInject will inject all fields to the pod
func (pi *PodInject) execute(ipd *InjectProcessData) admission.Response {
   log.Info("=============== PodInject ================")
   ipd.injectFileds.Inject(ipd.pod)
   // Pod注入完成後，新增sidecar.skywalking.apache.org/succeed=true註解
   ipd.injectFileds.injectSucceedAnnotation(&ipd.pod.Annotations)
   log.Info("inject successfully!")
   // 序列化Pod,返回給k8s
   return PatchReq(ipd.pod, ipd.req)
}

// Inject will do real injection
func (s *SidecarInjectField) Inject(pod *corev1.Pod) {
   log.Info(fmt.Sprintf("inject pod : %s", pod.GenerateName))
   // 將之前執行得到的InitContainer與pod設定的InitContainer合併在一起，也就是說pod initcontainer可以有多個
   if pod.Spec.InitContainers != nil {
      pod.Spec.InitContainers = append(pod.Spec.InitContainers, s.Initcontainer)
   } else {
      pod.Spec.InitContainers = []corev1.Container{s.Initcontainer}
   }

   // add volume to spec
   if pod.Spec.Volumes == nil {
      pod.Spec.Volumes = []corev1.Volume{}
   }
   pod.Spec.Volumes = append(pod.Spec.Volumes, s.SidecarVolume)
   if len(s.ConfigmapVolume.Name) > 0 && len(s.ConfigmapVolume.ConfigMap.Name) > 0 {
      pod.Spec.Volumes = append(pod.Spec.Volumes, s.ConfigmapVolume)
   }

   //選擇要注入的目標容器
   targetContainers := s.findInjectContainer(pod.Spec.Containers)

   //迴圈目標容器進行注入
   for i := range targetContainers {
      log.Info(fmt.Sprintf("inject container : %s", targetContainers[i].Name))
      if (*targetContainers[i]).VolumeMounts == nil {
         (*targetContainers[i]).VolumeMounts = []corev1.VolumeMount{}
      }

      // 注入voLume與configmap
      (*targetContainers[i]).VolumeMounts = append((*targetContainers[i]).VolumeMounts, s.SidecarVolumeMount)
      if len(s.ConfigmapVolumeMount.Name) > 0 && len(s.ConfigmapVolumeMount.MountPath) > 0 {
         (*targetContainers[i]).VolumeMounts = append((*targetContainers[i]).VolumeMounts, s.ConfigmapVolumeMount)
      }

      //java agent引數,其值為上面的JvmAgentConfigStr
      if (*targetContainers[i]).Env != nil {
         (*targetContainers[i]).Env = append((*targetContainers[i]).Env, s.Env)
      } else {
         (*targetContainers[i]).Env = []corev1.EnvVar{s.Env}
      }

      //注入環境變數，如果container本身存在,則忽略
      var envsTBA []corev1.EnvVar
      for j, envInject := range s.Envs {
         isExists := false
         for _, envExists := range targetContainers[i].Env {
            if strings.EqualFold(envExists.Name, envInject.Name) {
               isExists = true
               break
            }
         }
         if !isExists {
            envsTBA = append(envsTBA, s.Envs[j])
         }
      }
      if len(s.Envs) > 0 {
         (*targetContainers[i]).Env = append((*targetContainers[i]).Env, envsTBA...)
      }
   }
}

除了Pod注入，SWCK專案還有其它Operator, 包括Storage,OAP,UI,Adapter等，有興趣的話可自行探索。總體來說swck利用k8s的自定義資源以及自定義控制器，為skywalking部署到kubernetes提供了適配,使skywalking能夠快速部署到kubernetes這個基座上。

注意事項

SwAgent只能在業務空間起作用，不能在skywalking-swck-system生效

因為webhook觸發呼叫handler後，在查詢SwAgent時，只會查詢與Pod在一個名稱空間中的Swagent. 如果想將SwAgent放到skywalking-swck-system命令空間，需要修改operator

刪除資源時JavaAgent狀態中統計的注入的容器數量不變化

因為MutatingWebhookConfiguration只監聽了Pod的Create與Update事件。

偵錯需要問題

本地啟用webhook前提下無法啟動operator

因為啟動webhook時，需要在本地啟動webhook server，與k8s叢集通過https通訊, 本地需要新增tls.crt以及tls.key檔案。而這兩個檔案從k8s獲取。具體方法是檢視skywalking-swck-controller-manager使用到的secret

kubectl get secret skywalking-swck-controller-manager-cert -n skywalking-swck-system -o jsonpath='{.data.tls\.crt}'| base64 --decode > tls.crt
kubectl get secret skywalking-swck-controller-manager-cert -n skywalking-swck-system -o jsonpath='{.data.tls\.key}'| base64 --decode > tls.key