vscode探索一波Kubesphere后端架构

$ cat .vscode/launch.json


{
    // 使用 IntelliSense 了解相关属性。 
    // 悬停以查看现有属性的描述。
    // 欲了解更多信息，请访问: https://go.microsoft.com/fwlink/?linkid=830387
    "version": "0.2.0",
    "configurations": [
        {
            "name": "Launch Package",
            "type": "go",
            "request": "launch",
            "mode": "auto",
            "program": "${workspaceFolder}/cmd/ks-apiserver/apiserver.go"
        }
        
    ]
}

$ kubectl -n kubesphere-system get cm kubesphere-config  -oyaml
apiVersion: v1
data:
  kubesphere.yaml: |
    authentication:
      authenticateRateLimiterMaxTries: 10
      authenticateRateLimiterDuration: 10m0s
      loginHistoryRetentionPeriod: 168h
      maximumClockSkew: 10s
      multipleLogin: True
      kubectlImage: kubesphere/kubectl:v1.20.0
      jwtSecret: "Xtc8ZWUf9f3cJN89bglrTJhfUPMZR87d"
      oauthOptions:
        clients:
        - name: kubesphere
          secret: kubesphere
          redirectURIs:
          - '*'
    network:
      ippoolType: none
    monitoring:
      endpoint: http://prometheus-operated.kubesphere-monitoring-system.svc:9090
      enableGPUMonitoring: false
    gpu:
      kinds:
      - resourceName: nvidia.com/gpu
        resourceType: GPU
        default: True
    notification:
      endpoint: http://notification-manager-svc.kubesphere-monitoring-system.svc:19093




    kubeedge:
      endpoint: http://edge-watcher.kubeedge.svc/api/


    gateway:
      watchesPath: /var/helm-charts/watches.yaml
      namespace: kubesphere-controls-system
kind: ConfigMap
metadata:
  name: kubesphere-config
  namespace: kubesphere-system

$ cat ./cmd/ks-apiserver/kubesphere.yaml
kubernetes:
  kubeconfig: "/root/.kube/config"
  master: https://192.168.88.6:6443
  $qps: 1e+06
  burst: 1000000
authentication:
  authenticateRateLimiterMaxTries: 10
  authenticateRateLimiterDuration: 10m0s
  loginHistoryRetentionPeriod: 168h
  maximumClockSkew: 10s
  multipleLogin: True
  kubectlImage: kubesphere/kubectl:v1.20.0
  jwtSecret: "Xtc8ZWUf9f3cJN89bglrTJhfUPMZR87d"
  oauthOptions:
    clients:
    - name: kubesphere
      secret: kubesphere
      redirectURIs:
      - '*'
network:
  ippoolType: none
monitoring:
  endpoint: http://prometheus-operated.kubesphere-monitoring-system.svc:9090
  enableGPUMonitoring: false
gpu:
  kinds:
  - resourceName: nvidia.com/gpu
    resourceType: GPU
    default: True
notification:
  endpoint: http://notification-manager-svc.kubesphere-monitoring-system.svc:19093




kubeedge:
  endpoint: http://edge-watcher.kubeedge.svc/api/


gateway:
  watchesPath: /var/helm-charts/watches.yaml
  namespace: kubesphere-controls-system

// Load configuration from file
conf, err := apiserverconfig.TryLoadFromDisk()

viper.SetConfigName(defaultConfigurationName) // defaultConfigurationName = "kubesphere"


// AddConfigPath adds a path for Viper to search for the config file in.
// Can be called multiple times to define multiple search paths.
viper.AddConfigPath(defaultConfigurationPath) // defaultConfigurationPath = "/etc/kubesphere"


// Load from current working directory, only used for debugging
viper.AddConfigPath(".")


// Load from Environment variables
// E.g. if your prefix is "spf", the env registry will look for env variables that start with "SPF_".
viper.SetEnvPrefix("kubesphere")
viper.AutomaticEnv()
viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))


// 单步调试，这一步读取的文件路径是  jww.INFO.Println("Searching for config in ", v.configPaths)
// v.configPaths：["/etc/kubesphere","/root/go/src/kubesphere.io/kubesphere/cmd/ks-apiserver"]
if err := viper.ReadInConfig(); err != nil {
    if _, ok := err.(viper.ConfigFileNotFoundError); ok {
      return nil, err
    } else {
      return nil, fmt.Errorf("error parsing configuration file %s", err)
    }
}


conf := New() // 获取各个组件的默认配置


if err := viper.Unmarshal(conf); err != nil {
    return nil, err
}


return conf, nil

func Run(s *options.ServerRunOptions, ctx context.Context) error {
// NewAPIServer 通过给定的配置启动apiserver实例，绑定实例化的各个模块的client，绑定RuntimeClient和RuntimeCache
apiserver, err := s.NewAPIServer(ctx.Done())
if err != nil {
return err
}


// PrepareRun 主要是使用resful-go集成kapis, 也就是代理服务
err = apiserver.PrepareRun(ctx.Done())
if err != nil {
return nil
}


// Start listening, 开启监听
return apiserver.Run(ctx)
}

...
server := &http.Server{
    Addr: fmt.Sprintf(":%d", s.GenericServerRunOptions.InsecurePort),
}


if s.GenericServerRunOptions.SecurePort != 0 {
    certificate, err := tls.LoadX509KeyPair(s.GenericServerRunOptions.TlsCertFile, s.GenericServerRunOptions.TlsPrivateKey)
    if err != nil {
        return nil, err
    }


    server.TLSConfig = &tls.Config{
        Certificates: []tls.Certificate{certificate},
    }
    server.Addr = fmt.Sprintf(":%d", s.GenericServerRunOptions.SecurePort)
}
    
sch := scheme.Scheme
if err := apis.AddToScheme(sch); err != nil {
    klog.Fatalf("unable add APIs to scheme: %v", err)
}
...

// AddToSchemes may be used to add all resources defined in the project to a Scheme
var AddToSchemes runtime.SchemeBuilder


// AddToScheme adds all Resources to the Scheme
func AddToScheme(s *runtime.Scheme) error {
return AddToSchemes.AddToScheme(s)
}

$ cat pkg/apis/addtoscheme_dashboard_v1alpha2.gopackage apisimport monitoringdashboardv1alpha2 "kubesphere.io/monitoring-dashboard/api/v1alpha2"func init() {  AddToSchemes = append(AddToSchemes, monitoringdashboardv1alpha2.SchemeBuilder.AddToScheme)}

func (s *APIServer) PrepareRun(stopCh <-chan struct{}) error {  s.container = restful.NewContainer()  / 添加请求Request日志拦截器  s.container.Filter(logRequestAndResponse)  s.container.Router(restful.CurlyRouter{})    / RecoverHandler changes the default function (logStackOnRecover) to be called  / when a panic is detected. DoNotRecover must be have its default value (=false).  s.container.RecoverHandler(func(panicReason interface{}, httpWriter http.ResponseWriter) {    logStackOnRecover(panicReason, httpWriter)  })    / 注册、绑定kapi路由和回调函数，发生在所有informers启动之前  s.installKubeSphereAPIs()    / 注册metrics指标: ks_server_request_total、ks_server_request_duration_seconds  / 绑定API kapis/metrics以及handle， 主要是调用prometheus collector集成，在pkg/utils/metrics/metrics.go  s.installMetricsAPI()  / 过滤无效监控请求  s.container.Filter(monitorRequest)  for _, ws := range s.container.RegisteredWebServices() {    klog.V(2).Infof("%s", ws.RootPath())  }    / 将container绑定给s.Server.Handler  s.Server.Handler = s.container    / 添加k8s api资源、auditing资源、登录验证、凭据验证等各个调用链的拦截器  s.buildHandlerChain(stopCh)  return nil}

// 调用各api组的AddToContainer方法来向container注册api:
urlruntime.Must(monitoringv1alpha3.AddToContainer(s.container, s.KubernetesClient.Kubernetes(), s.MonitoringClient, s.MetricsClient, s.InformerFactory, s.KubernetesClient.KubeSphere(), s.Config.OpenPitrixOptions))
// 详细来说，各个组件实现的AddToContainer方法则实现了带组和版本信息子路由和handler的绑定：// 首先给router绑定/kapis的父router。ws := runtime.NewWebService(GroupVersion)// 给子路由绑定回调函数
ws.Route(ws.GET("/kubesphere")
.To(h.handleKubeSphereMetricsQuery)
.Doc("Get platform-level metric data.")
.Metadata(restfulspec.KeyOpenAPITags, []string{constants.KubeSphereMetricsTag})
.Writes(model.Metrics{})
.Returns(http.StatusOK, respOK, model.Metrics{}))
.Produces(restful.MIME_JSON)

$ curl "localhost:9090/kapis/iam.kubesphere.io/v1alpha2/users"
$ curl "localhost:9090/kapis/iam.kubesphere.io/v1alpha2/clustermembers"
$ curl "localhost:9090/kapis/iam.kubesphere.io/v1alpha2/users/admin/globalroles"...

$ curl -X POST -H "Content-Type: application/json" -d '{"username":"admin","password":"P@88w0rd"}' "http://localhost:9090/kapis/iam.kubesphere.io/v1alpha2/login"

func AddToContainer(container *restful.Container, endpoint string) error {  proxy, err := generic.NewGenericProxy(endpoint, GroupVersion.Group, GroupVersion.Version)  { 
   if err != nil { 
      return nil  
   }  
   return proxy.AddToContainer(container)
 }

func (h *handler) ListResource(req *restful.Request, resp *restful.Response) {  
   // 租户或用户的名称  
   user := req.PathParameter("user") 
   // 资源类型，configs/recievers/secrets  
   resource := req.PathParameter("resources") 
   // 通知渠道 dingtalk/slack/email/webhook/wechat 
   subresource := req.QueryParameter("type") 
   q := query.ParseQueryParameter(req)
   if !h.operator.IsKnownResource(resource, subresource) {   
      api.HandleBadRequest(resp, req, servererr.New("unknown resource type %s/%s", resource, subresource))
      return  
   }  
   objs, err := h.operator.List(user, resource, subresource, q)
   handleResponse(req, resp, objs, err)
 }

// List objects
func (o *operator) List(user, resource, subresource string, q *query.Query) (*api.ListResult, error) { 
 if len(q.LabelSelector) > 0 {
     q.LabelSelector = q.LabelSelector + ","  
 }
 filter := ""  // 如果没有给定租户的名称，则获取全局的对象
 if user == "" {   
      if isConfig(o.GetObject(resource)) {   
      // type=default对config资源来说是全局的 
           filter = "type=default"   
      } else {      
        // type=global对receiever资源来说是全局的     
        filter = "type=global"   
      } 
 } else {  
 // 否则就给过滤器绑定租户名称   
    filter = "type=tenant,user=" + user  
  }  
  // 组装过滤标签 
    q.LabelSelector = q.LabelSelector + filter
     ... 
  // 通过过滤标签获取cluster或者namespace下的指定资源
    res, err := o.resourceGetter.List(resource, ns, q) 
    if err != nil {  
        return nil, err 
    } 
      if subresource == "" || resource == Secret { 
         return res, nil
      }  
      
      results := &api.ListResult{}  
      ...
}

func (h handler) handleNamedMetricsQuery(resp *restful.Response, q queryOptions) {
  var res model.Metrics


  var metrics []string
  // q.namedMetrics 是一组按照监控指标级别分类好的拥有promsql expr定义的完整指标名数组
  // 监控指标级别分类是根据 monitoring.Levelxxx在上一个栈里细分的，i.e: monitoring.LevelPod
  for _, metric := range q.namedMetrics {
    if strings.HasPrefix(metric, model.MetricMeterPrefix) {
      // skip meter metric
      continue
    }
    // 根据请求参数中的指标名来过滤
    ok, _ := regexp.MatchString(q.metricFilter, metric)
    if ok {
      metrics = append(metrics, metric)
    }
  }
  if len(metrics) == 0 {
    resp.WriteAsJson(res)
    return
  }
  
  // 判断是否是范围查询还是实时查询，继续调用相关函数
  // 主要还是用prometheus client去查询promsql, 边缘节点的指标目前通过metrics server来查询
  if q.isRangeQuery() {
    res = h.mo.GetNamedMetricsOverTime(metrics, q.start, q.end, q.step, q.option)
  } else {
    res = h.mo.GetNamedMetrics(metrics, q.time, q.option)
    if q.shouldSort() {
      res = *res.Sort(q.target, q.order, q.identifier).Page(q.page, q.limit)
    }
  }
  resp.WriteAsJson(res)
}

func (mo monitoringOperator) GetNamedMetricsOverTime(metrics []string, start, end time.Time, step time.Duration, opt monitoring.QueryOption) Metrics {
    // 获取prometheus client查询结果，主要使用sync.WaitGroup并发查询，每个指标启动一个goroutine，最后将结果和并返回
  ress := mo.prometheus.GetNamedMetricsOverTime(metrics, start, end, step, opt)
  // 如果metrics-server激活了
  if mo.metricsserver != nil {


    //合并边缘节点数据
    edgeMetrics := make(map[string]monitoring.MetricData)


    for i, ressMetric := range ress {
      metricName := ressMetric.MetricName
      ressMetricValues := ressMetric.MetricData.MetricValues
      if len(ressMetricValues) == 0 {
        // this metric has no prometheus metrics data
        if len(edgeMetrics) == 0 {
          // start to request monintoring metricsApi data
          mr := mo.metricsserver.GetNamedMetricsOverTime(metrics, start, end, step, opt)
          for _, mrMetric := range mr {
            edgeMetrics[mrMetric.MetricName] = mrMetric.MetricData
          }
        }
        if val, ok := edgeMetrics[metricName]; ok {
          ress[i].MetricData.MetricValues = append(ress[i].MetricData.MetricValues, val.MetricValues...)
        }
      }
    }
  }


  return Metrics{Results: ress}
}

func (h *tenantHandler) QueryLogs(req *restful.Request, resp *restful.Response) {
    // 查询上下文中携带的租户信息
  user, ok := request.UserFrom(req.Request.Context())
  if !ok {
    err := fmt.Errorf("cannot obtain user info")
    klog.Errorln(err)
    api.HandleForbidden(resp, req, err)
    return
  }
  // 解析查询的参数，比如确定属于哪个ns/workload/pod/container的查询、时间段，是否为柱状查询等
  queryParam, err := loggingv1alpha2.ParseQueryParameter(req)
  if err != nil {
    klog.Errorln(err)
    api.HandleInternalError(resp, req, err)
    return
  }
  // 导出数据
  if queryParam.Operation == loggingv1alpha2.OperationExport {
    resp.Header().Set(restful.HEADER_ContentType, "text/plain")
    resp.Header().Set("Content-Disposition", "attachment")
    // 验证账号是否有权限
    // admin账号可以导出所有ns的日志，租户只能导出本ns的日志
    // 组装loggingclient进行日志导出
    err := h.tenant.ExportLogs(user, queryParam, resp)
    if err != nil {
      klog.Errorln(err)
      api.HandleInternalError(resp, req, err)
      return
    }
  } else {
    // 验证账号是否有权限
    // admin账号可以查看所有ns的日志，租户只能查看本ns的日志
    // 组装loggingclient进行日志返回
    result, err := h.tenant.QueryLogs(user, queryParam)
    if err != nil {
      klog.Errorln(err)
      api.HandleInternalError(resp, req, err)
      return
    }
    resp.WriteAsJson(result)
  }
}

// Authorizer makes an authorization decision based on information gained by making
// zero or more calls to methods of the Attributes interface.  It returns nil when an action is
// authorized, otherwise it returns an error.
type Authorizer interface {
  Authorize(a Attributes) (authorized Decision, reason string, err error)
}

var authorizers authorizer.Authorizer


// 默认是RBAC，可以通过authorization.mode在启动配置文件修改
switch s.Config.AuthorizationOptions.Mode { 
case authorizationoptions.AlwaysAllow:
  authorizers = authorizerfactory.NewAlwaysAllowAuthorizer()
case authorizationoptions.AlwaysDeny:
  authorizers = authorizerfactory.NewAlwaysDenyAuthorizer()
default:
  fallthrough
case authorizationoptions.RBAC:
  excludedPaths := []string{"/oauth/*", "/kapis/config.kubesphere.io/*", "/kapis/version", "/kapis/metrics"}
  pathAuthorizer, _ := path.NewAuthorizer(excludedPaths)
  amOperator := am.NewReadOnlyOperator(s.InformerFactory)
  authorizers = unionauthorizer.New(pathAuthorizer, rbac.NewRBACAuthorizer(amOperator))
}