Add support for StatefulSets

apply deislabs#43

With this change, Osiris can now manage both Deployments and StatefulSets.

I introduced a new service annotation `osiris.deislabs.io/statefulset` to link a service with its statefulset, otherwise everything works the same as with Deployments.

Internally, I tried to avoid duplicated code, and introduced a `kind` parameter to make a distinction between deployments and statefulsets.

Author: vbehar

README.md

@@ -29,11 +29,11 @@ Various types of Kubernetes resources can be Osiris-enabled using an annotation.
 Osiris-enabled pods are automatically instrumented with a __metrics-collecting
 proxy__ deployed as a sidecar container.
 
-Osiris-enabled deployments (if _already_ scaled to a configurable minimum number
-of replicas-- one by default) automatically have metrics from their pods
-continuously scraped and analyzed by the __zeroscaler__ component. When the
-aggregated metrics reveal that all of the deployment's pods are idling, the
-zeroscaler scales the deployment to zero replicas.
+Osiris-enabled deployments or statefulSets (if _already_ scaled to a configurable
+minimum number of replicas-- one by default) automatically have metrics from
+their pods continuously scraped and analyzed by the __zeroscaler__ component.
+When the aggregated metrics reveal that all of the deployment's pods are idling,
+the zeroscaler scales the deployment to zero replicas.
 
 Under normal circumstances, scaling a deployment to zero replicas poses a
 problem: any services that select pods from that deployment (and only that
@@ -175,14 +175,14 @@ spec:
 
 Most of Osiris configuration is done with Kubernetes annotations - as seen in the Usage section.
 
-#### Deployment Annotations
+#### Deployment & StatefulSet Annotations
 
-The following table lists the supported annotations for Kubernetes `Deployments` and their default values.
+The following table lists the supported annotations for Kubernetes `Deployments` and `StatefulSets`, and their default values.
 
 | Annotation | Description | Default |
 | ---------- | ----------- | ------- |
-| `osiris.dm.gg/enabled` | Enable the zeroscaler component to scrape and analyze metrics from the deployment's pods and scale the deployment to zero when idle. Allowed values: `y`, `yes`, `true`, `on`, `1`. | _no value_ (= disabled) |
-| `osiris.dm.gg/minReplicas` | The minimum number of replicas to set on the deployment when Osiris will scale up. If you set `2`, Osiris will scale the deployment from `0` to `2` replicas directly. Osiris won't collect metrics from deployments which have more than `minReplicas` replicas - to avoid useless collections of metrics. | `1` |
+| `osiris.dm.gg/enabled` | Enable the zeroscaler component to scrape and analyze metrics from the deployment's or statefulSet's pods and scale the deployment/statefulSet to zero when idle. Allowed values: `y`, `yes`, `true`, `on`, `1`. | _no value_ (= disabled) |
+| `osiris.dm.gg/minReplicas` | The minimum number of replicas to set on the deployment/statefulSet when Osiris will scale up. If you set `2`, Osiris will scale the deployment/statefulSet from `0` to `2` replicas directly. Osiris won't collect metrics from deployments/statefulSets which have more than `minReplicas` replicas - to avoid useless collections of metrics. | `1` |
 | `osiris.dm.gg/metricsCheckInterval` | The interval in which Osiris would repeatedly track the pod http request metrics. The value is the number of seconds of the interval. Note that this value override the global value defined by the `zeroscaler.metricsCheckInterval` Helm value. | _value of the `zeroscaler.metricsCheckInterval` Helm value_ |
 
 #### Pod Annotations
@@ -202,6 +202,7 @@ The following table lists the supported annotations for Kubernetes `Services` an
 | ---------- | ----------- | ------- |
 | `osiris.dm.gg/enabled` | Enable this service's endpoints to be managed by the Osiris endpoints controller. Allowed values: `y`, `yes`, `true`, `on`, `1`. | _no value_ (= disabled) |
 | `osiris.dm.gg/deployment` | Name of the deployment which is behind this service. This is _required_ to map the service with its deployment. | _no value_ |
+| `osiris.dm.gg/statefulset` | Name of the statefulSet which is behind this service. This is _required_ to map the service with its statefulSet. | _no value_ |
 | `osiris.dm.gg/loadBalancerHostname` | Map requests coming from a specific hostname to this service. Note that if you have multiple hostnames, you can set them with different annotations, using `osiris.dm.gg/loadBalancerHostname-1`, `osiris.dm.gg/loadBalancerHostname-2`, ... | _no value_ |
 | `osiris.dm.gg/ingressHostname` | Map requests coming from a specific hostname to this service. If you use an ingress in front of your service, this is required to create a link between the ingress and the service. Note that if you have multiple hostnames, you can set them with different annotations, using `osiris.dm.gg/ingressHostname-1`, `osiris.dm.gg/ingressHostname-2`, ... | _no value_ |
 | `osiris.dm.gg/ingressDefaultPort` | Custom service port when the request comes from an ingress. Default behaviour if there are more than 1 port on the service, is to look for a port named `http`, and fallback to the port `80`. Set this if you have multiple ports and using a non-standard port with a non-standard name. | _no value_ |

charts/osiris/templates/cluster-role.yaml

@@ -32,6 +32,7 @@ rules:
   - apps
   resources:
   - deployments
+  - statefulsets
   verbs:
   - get
   - list

pkg/deployments/activator/activating.go

@@ -15,24 +15,24 @@ import (
 func (a *activator) activateDeployment(
 	ctx context.Context,
 	app *app,
-) (*deploymentActivation, error) {
+) (*appActivation, error) {
 	deploymentsClient := a.kubeClient.AppsV1().Deployments(app.namespace)
 	deployment, err := deploymentsClient.Get(
 		ctx,
-		app.deploymentName,
+		app.name,
 		metav1.GetOptions{},
 	)
 	if err != nil {
 		return nil, err
 	}
-	da := &deploymentActivation{
+	da := &appActivation{
 		readyAppPodIPs: map[string]struct{}{},
 		successCh:      make(chan struct{}),
 		timeoutCh:      make(chan struct{}),
 	}
 	glog.Infof(
 		"Activating deployment %s in namespace %s",
-		app.deploymentName,
+		app.name,
 		app.namespace,
 	)
 	go da.watchForCompletion(
@@ -55,7 +55,58 @@ func (a *activator) activateDeployment(
 	patchesBytes, _ := json.Marshal(patches)
 	_, err = deploymentsClient.Patch(
 		ctx,
-		app.deploymentName,
+		app.name,
+		k8s_types.JSONPatchType,
+		patchesBytes,
+		metav1.PatchOptions{},
+	)
+	return da, err
+}
+
+func (a *activator) activateStatefulSet(
+	ctx context.Context,
+	app *app,
+) (*appActivation, error) {
+	statefulSetsClient := a.kubeClient.AppsV1().StatefulSets(app.namespace)
+	statefulSet, err := statefulSetsClient.Get(
+		ctx,
+		app.name,
+		metav1.GetOptions{},
+	)
+	if err != nil {
+		return nil, err
+	}
+	da := &appActivation{
+		readyAppPodIPs: map[string]struct{}{},
+		successCh:      make(chan struct{}),
+		timeoutCh:      make(chan struct{}),
+	}
+	glog.Infof(
+		"Activating statefulSet %s in namespace %s",
+		app.name,
+		app.namespace,
+	)
+	go da.watchForCompletion(
+		a.kubeClient,
+		app,
+		labels.Set(statefulSet.Spec.Selector.MatchLabels).AsSelector(),
+	)
+	if statefulSet.Spec.Replicas == nil || *statefulSet.Spec.Replicas > 0 {
+		// We don't need to do this, as it turns out! Scaling is either already
+		// in progress-- perhaps initiated by another process-- or may even be
+		// completed already. Just return dr and allow the caller to move on to
+		// verifying / waiting for this activation to be complete.
+		return da, nil
+	}
+	patches := []kubernetes.PatchOperation{{
+		Op:    "replace",
+		Path:  "/spec/replicas",
+		Value: kubernetes.GetMinReplicas(statefulSet.Annotations, 1),
+	}}
+	patchesBytes, _ := json.Marshal(patches)
+	_, err = statefulSetsClient.Patch(
+		ctx,
+		app.name,
 		k8s_types.JSONPatchType,
 		patchesBytes,
 		metav1.PatchOptions{},

pkg/deployments/activator/activator.go

@@ -21,16 +21,16 @@ type Activator interface {
 }
 
 type activator struct {
-	kubeClient                kubernetes.Interface
-	servicesInformer          cache.SharedIndexInformer
-	nodeInformer              cache.SharedIndexInformer
-	services                  map[string]*corev1.Service
-	nodeAddresses             map[string]struct{}
-	appsByHost                map[string]*app
-	indicesLock               sync.RWMutex
-	deploymentActivations     map[string]*deploymentActivation
-	deploymentActivationsLock sync.Mutex
-	srv                       *http.Server
+	kubeClient         kubernetes.Interface
+	servicesInformer   cache.SharedIndexInformer
+	nodeInformer       cache.SharedIndexInformer
+	services           map[string]*corev1.Service
+	nodeAddresses      map[string]struct{}
+	appsByHost         map[string]*app
+	indicesLock        sync.RWMutex
+	appActivations     map[string]*appActivation
+	appActivationsLock sync.Mutex
+	srv                *http.Server
 }
 
 func NewActivator(kubeClient kubernetes.Interface) Activator {
@@ -56,8 +56,8 @@ func NewActivator(kubeClient kubernetes.Interface) Activator {
 			Addr:    fmt.Sprintf(":%d", port),
 			Handler: mux,
 		},
-		appsByHost:            map[string]*app{},
-		deploymentActivations: map[string]*deploymentActivation{},
+		appsByHost:     map[string]*app{},
+		appActivations: map[string]*appActivation{},
 	}
 	a.servicesInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: a.syncService,
@@ -106,7 +106,7 @@ func (a *activator) syncService(obj interface{}) {
 	a.indicesLock.Lock()
 	defer a.indicesLock.Unlock()
 	svc := obj.(*corev1.Service)
-	svcKey := getKey(svc.Namespace, svc.Name)
+	svcKey := getKey(svc.Namespace, "Service", svc.Name)
 	if k8s.ResourceIsOsirisEnabled(svc.Annotations) {
 		a.services[svcKey] = svc
 	} else {
@@ -119,7 +119,7 @@ func (a *activator) syncDeletedService(obj interface{}) {
 	a.indicesLock.Lock()
 	defer a.indicesLock.Unlock()
 	svc := obj.(*corev1.Service)
-	svcKey := getKey(svc.Namespace, svc.Name)
+	svcKey := getKey(svc.Namespace, "Service", svc.Name)
 	delete(a.services, svcKey)
 	a.updateIndex()
 }

pkg/deployments/activator/app.go

@@ -5,10 +5,18 @@ import (
 	"net/url"
 )
 
+type appKind string
+
+const (
+	appKindDeployment  appKind = "Deployment"
+	appKindStatefulSet appKind = "StatefulSet"
+)
+
 type app struct {
 	namespace           string
 	serviceName         string
-	deploymentName      string
+	name                string
+	kind                appKind
 	targetURL           *url.URL
 	proxyRequestHandler *httputil.ReverseProxy
 }

pkg/deployments/activator/deployment_activation.go renamed to pkg/deployments/activator/app_activation.go

@@ -15,34 +15,34 @@ import (
 	k8s "github.com/dailymotion/osiris/pkg/kubernetes"
 )
 
-type deploymentActivation struct {
+type appActivation struct {
 	readyAppPodIPs map[string]struct{}
 	endpoints      *corev1.Endpoints
 	lock           sync.Mutex
 	successCh      chan struct{}
 	timeoutCh      chan struct{}
 }
 
-func (d *deploymentActivation) watchForCompletion(
+func (a *appActivation) watchForCompletion(
 	kubeClient kubernetes.Interface,
 	app *app,
 	appPodSelector labels.Selector,
 ) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	// Watch the pods managed by this deployment
+	// Watch the pods managed by this deployment/statefulSet
 	podsInformer := k8s.PodsIndexInformer(
 		kubeClient,
 		app.namespace,
 		nil,
 		appPodSelector,
 	)
 	podsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: d.syncPod,
+		AddFunc: a.syncPod,
 		UpdateFunc: func(_, newObj interface{}) {
-			d.syncPod(newObj)
+			a.syncPod(newObj)
 		},
-		DeleteFunc: d.syncPod,
+		DeleteFunc: a.syncPod,
 	})
 	// Watch the corresponding endpoints resource for this service
 	endpointsInformer := k8s.EndpointsIndexInformer(
@@ -55,9 +55,9 @@ func (d *deploymentActivation) watchForCompletion(
 		nil,
 	)
 	endpointsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: d.syncEndpoints,
+		AddFunc: a.syncEndpoints,
 		UpdateFunc: func(_, newObj interface{}) {
-			d.syncEndpoints(newObj)
+			a.syncEndpoints(newObj)
 		},
 	})
 	go podsInformer.Run(ctx.Done())
@@ -66,23 +66,24 @@ func (d *deploymentActivation) watchForCompletion(
 	defer timer.Stop()
 	for {
 		select {
-		case <-d.successCh:
+		case <-a.successCh:
 			return
 		case <-timer.C:
 			glog.Errorf(
-				"Activation of deployment %s in namespace %s timed out",
-				app.deploymentName,
+				"Activation of %s %s in namespace %s timed out",
+				app.kind,
+				app.name,
 				app.namespace,
 			)
-			close(d.timeoutCh)
+			close(a.timeoutCh)
 			return
 		}
 	}
 }
 
-func (d *deploymentActivation) syncPod(obj interface{}) {
-	d.lock.Lock()
-	defer d.lock.Unlock()
+func (a *appActivation) syncPod(obj interface{}) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
 	pod := obj.(*corev1.Pod)
 	var ready bool
 	for _, condition := range pod.Status.Conditions {
@@ -95,27 +96,27 @@ func (d *deploymentActivation) syncPod(obj interface{}) {
 	}
 	// Keep track of which pods are ready
 	if ready {
-		d.readyAppPodIPs[pod.Status.PodIP] = struct{}{}
+		a.readyAppPodIPs[pod.Status.PodIP] = struct{}{}
 	} else {
-		delete(d.readyAppPodIPs, pod.Status.PodIP)
+		delete(a.readyAppPodIPs, pod.Status.PodIP)
 	}
-	d.checkActivationComplete()
+	a.checkActivationComplete()
 }
 
-func (d *deploymentActivation) syncEndpoints(obj interface{}) {
-	d.lock.Lock()
-	defer d.lock.Unlock()
-	d.endpoints = obj.(*corev1.Endpoints)
-	d.checkActivationComplete()
+func (a *appActivation) syncEndpoints(obj interface{}) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.endpoints = obj.(*corev1.Endpoints)
+	a.checkActivationComplete()
 }
 
-func (d *deploymentActivation) checkActivationComplete() {
-	if d.endpoints != nil {
-		for _, subset := range d.endpoints.Subsets {
+func (a *appActivation) checkActivationComplete() {
+	if a.endpoints != nil {
+		for _, subset := range a.endpoints.Subsets {
 			for _, address := range subset.Addresses {
-				if _, ok := d.readyAppPodIPs[address.IP]; ok {
+				if _, ok := a.readyAppPodIPs[address.IP]; ok {
 					glog.Infof("App pod with ip %s is in service", address.IP)
-					close(d.successCh)
+					close(a.successCh)
 					return
 				}
 			}