ElasticSearch Operator 工作原理浅析

kubectl apply -f https://download.elastic.co/downloads/eck/1.1.2/all-in-one.yaml

k get crd | grep elastic
apmservers.apm.k8s.elastic.co                  2020-05-10T08:02:15Z
elasticsearches.elasticsearch.k8s.elastic.co   2020-05-10T08:02:15Z
kibanas.kibana.k8s.elastic.co                  2020-05-10T08:02:15Z

apiVersion: v1
kind: Namespace
metadata:
  creationTimestamp: null
  name: elastic-stack
---
apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: elasticsearch
  namespace: elastic-stack
spec:
  version: 7.6.2
  nodeSets:
  - name: default
    count: 1
    config:
      node.master: true
      node.data: true
      node.ingest: true
      node.store.allow_mmap: false
---
apiVersion: kibana.k8s.elastic.co/v1
kind: Kibana
metadata:
  name: kibana
  namespace: elastic-stack
spec:
  version: 7.6.2
  count: 1
  elasticsearchRef:
    name: elasticsearch
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: es-db-0
  labels:
    type: local
spec:
  volumeMode: Filesystem
  persistentVolumeReclaimPolicy: Retain
  capacity:
    storage: 5Gi
  accessModes:
    - ReadWriteOnce
  hostPath:
    path: "/data/elastic/db-0"
---
apiVersion: v1
kind: PersistentVolume
metadata:
  name: es-db-1
  labels:
    type: local
spec:
  volumeMode: Filesystem
  persistentVolumeReclaimPolicy: Retain
  capacity:
    storage: 5Gi
  accessModes:
    - ReadWriteOnce
  hostPath:
    path: "/data/elastic/db-1"

type validation func(*Elasticsearch) field.ErrorList

// validations are the validation funcs that apply to creates or updates
var validations = []validation{
  noUnknownFields,
  validName,
  hasMaster,
  supportedVersion,
  validSanIP,
}

type updateValidation func(*Elasticsearch, *Elasticsearch) field.ErrorList

// updateValidations are the validation funcs that only apply to updates
var updateValidations = []updateValidation{
  noDowngrades,
  validUpgradePath,
  pvcModification,
}

$ k -n elastic-stack get svc
NAME                         TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)    AGE
elasticsearch-es-http        ClusterIP   10.96.42.27     <none>        9200/TCP   103d
elasticsearch-es-transport   ClusterIP   None            <none>        9300/TCP   103d

func HandleDownscale(
  downscaleCtx downscaleContext,
  expectedStatefulSets sset.StatefulSetList,
  actualStatefulSets sset.StatefulSetList,
) *reconciler.Results {
  results := &reconciler.Results{}

  // make sure we only downscale nodes we're allowed to
  downscaleState, err := newDownscaleState(downscaleCtx.k8sClient, downscaleCtx.es)
  if err != nil {
    return results.WithError(err)
  }

  // compute the list of StatefulSet downscales and deletions to perform
  downscales, deletions := calculateDownscales(*downscaleState, expectedStatefulSets, actualStatefulSets)

  // remove actual StatefulSets that should not exist anymore (already downscaled to 0 in the past)
  // this is safe thanks to expectations: we're sure 0 actual replicas means 0 corresponding pods exist
  if err := deleteStatefulSets(deletions, downscaleCtx.k8sClient, downscaleCtx.es); err != nil {
    return results.WithError(err)
  }

  // migrate data away from nodes that should be removed
  // if leavingNodes is empty, it clears any existing settings
  leavingNodes := leavingNodeNames(downscales)
  if err := migration.MigrateData(downscaleCtx.parentCtx, downscaleCtx.es, downscaleCtx.esClient, leavingNodes); err != nil {
    return results.WithError(err)
  }

  for _, downscale := range downscales {
    // attempt the StatefulSet downscale (may or may not remove nodes)
    requeue, err := attemptDownscale(downscaleCtx, downscale, actualStatefulSets)
    if err != nil {
      return results.WithError(err)
    }
    if requeue {
      // retry downscaling this statefulset later
      results.WithResult(defaultRequeue)
    }
  }

  return results
}

// healthChangeListener returns an OnObservation listener that feeds a generic
// event when a cluster's observed health has changed.
func healthChangeListener(reconciliation chan event.GenericEvent) OnObservation {
  return func(cluster types.NamespacedName, previous State, new State) {
    // no-op if health hasn't change
    if !hasHealthChanged(previous, new) {
      return
    }

    // trigger a reconciliation event for that cluster
    evt := event.GenericEvent{
      Meta: &metav1.ObjectMeta{
        Namespace: cluster.Namespace,
        Name:      cluster.Name,
      },
    }
    reconciliation <- evt
  }
}

// Controller implements a Kubernetes API.  A Controller manages a work queue fed reconcile.Requests
// from source.Sources.  Work is performed through the reconcile.Reconciler for each enqueued item.
// Work typically is reads and writes Kubernetes objects to make the system state match the state specified
// in the object Spec.
type Controller interface {
  // Reconciler is called to reconcile an object by Namespace/Name
  reconcile.Reconciler

  // Watch takes events provided by a Source and uses the EventHandler to
  // enqueue reconcile.Requests in response to the events.
  //
  // Watch may be provided one or more Predicates to filter events before
  // they are given to the EventHandler.  Events will be passed to the
  // EventHandler if all provided Predicates evaluate to true.
  Watch(src source.Source, eventhandler handler.EventHandler, predicates ...predicate.Predicate) error

  // Start starts the controller.  Start blocks until stop is closed or a
  // controller has an error starting.
  Start(stop <-chan struct{}) error
}

// License models the Elasticsearch license applied to a cluster. Signature will be empty on reads. IssueDate,  ExpiryTime and Status can be empty on writes.
type License struct {
  Status             string     `json:"status,omitempty"`
  UID                string     `json:"uid"`
  Type               string     `json:"type"`
  IssueDate          *time.Time `json:"issue_date,omitempty"`
  IssueDateInMillis  int64      `json:"issue_date_in_millis"`
  ExpiryDate         *time.Time `json:"expiry_date,omitempty"`
  ExpiryDateInMillis int64      `json:"expiry_date_in_millis"`
  MaxNodes           int        `json:"max_nodes,omitempty"`
  MaxResourceUnits   int        `json:"max_resource_units,omitempty"`
  IssuedTo           string     `json:"issued_to"`
  Issuer             string     `json:"issuer"`
  StartDateInMillis  int64      `json:"start_date_in_millis"`
  Signature          string     `json:"signature,omitempty"`
}

type ElasticsearchLicense struct {
  License client.License `json:"license"`
}

type EnterpriseLicense struct {
  License LicenseSpec `json:"license"`
}

type LicenseSpec struct {
  Status             string                 `json:"status,omitempty"`
  UID                string                 `json:"uid"`
  Type               OperatorLicenseType    `json:"type"`
  IssueDate          *time.Time             `json:"issue_date,omitempty"`
  IssueDateInMillis  int64                  `json:"issue_date_in_millis"`
  ExpiryDate         *time.Time             `json:"expiry_date,omitempty"`
  ExpiryDateInMillis int64                  `json:"expiry_date_in_millis"`
  MaxInstances       int                    `json:"max_instances,omitempty"`
  MaxResourceUnits   int                    `json:"max_resource_units,omitempty"`
  IssuedTo           string                 `json:"issued_to"`
  Issuer             string                 `json:"issuer"`
  StartDateInMillis  int64                  `json:"start_date_in_millis"`
  Signature          string                 `json:"signature,omitempty"`
  ClusterLicenses    []ElasticsearchLicense `json:"cluster_licenses"`
  Version            int                    // not marshalled but part of the signature
}