Merge pull request influxdata#1011 from influxdata/nc-k8s-replicasets

Add support for scaling replicasets too

k8s_autoscale.go

@@ -34,7 +34,6 @@ type K8sAutoscaleNode struct {
 
 	increaseCount      *expvar.Int
 	decreaseCount      *expvar.Int
-	errorsCount        *expvar.Int
 	cooldownDropsCount *expvar.Int
 
 	min int
@@ -70,19 +69,20 @@ func newK8sAutoscaleNode(et *ExecutingTask, n *pipeline.K8sAutoscaleNode, l *log
 func (k *K8sAutoscaleNode) runAutoscale([]byte) error {
 	k.increaseCount = &expvar.Int{}
 	k.decreaseCount = &expvar.Int{}
-	k.errorsCount = &expvar.Int{}
+	errorsCount := &expvar.Int{}
 	k.cooldownDropsCount = &expvar.Int{}
 
 	k.statMap.Set(statsK8sIncreaseEventsCount, k.increaseCount)
 	k.statMap.Set(statsK8sDecreaseEventsCount, k.decreaseCount)
-	k.statMap.Set(statsK8sErrorsCount, k.errorsCount)
+	k.statMap.Set(statsK8sErrorsCount, errorsCount)
 	k.statMap.Set(statsK8sCooldownDropsCount, k.cooldownDropsCount)
 
 	switch k.Wants() {
 	case pipeline.StreamEdge:
 		for p, ok := k.ins[0].NextPoint(); ok; p, ok = k.ins[0].NextPoint() {
 			k.timer.Start()
 			if np, err := k.handlePoint(p.Name, p.Group, p.Dimensions, p.Time, p.Fields, p.Tags); err != nil {
+				errorsCount.Add(1)
 				k.logger.Println("E!", err)
 			} else if np.Name != "" {
 				k.timer.Pause()
@@ -101,6 +101,7 @@ func (k *K8sAutoscaleNode) runAutoscale([]byte) error {
 			k.timer.Start()
 			for _, p := range b.Points {
 				if np, err := k.handlePoint(b.Name, b.Group, b.PointDimensions(), p.Time, p.Fields, p.Tags); err != nil {
+					errorsCount.Add(1)
 					k.logger.Println("E!", err)
 				} else if np.Name != "" {
 					k.timer.Pause()
@@ -143,7 +144,6 @@ func (k *K8sAutoscaleNode) handlePoint(streamName string, group models.GroupID,
 		// If we haven't seen this resource before, get its state
 		scale, err := k.getResource(namespace, kind, name)
 		if err != nil {
-			k.errorsCount.Add(1)
 			return models.Point{}, errors.Wrapf(err, "could not determine initial scale for %s/%s/%s", namespace, kind, name)
 		}
 		state.current = int(scale.Spec.Replicas)
@@ -205,7 +205,6 @@ func (k *K8sAutoscaleNode) handlePoint(streamName string, group models.GroupID,
 
 	// We have a valid event to apply
 	if err := k.applyEvent(e); err != nil {
-		k.errorsCount.Add(1)
 		return models.Point{}, errors.Wrap(err, "failed to apply scaling event")
 	}
 

pipeline/k8s_autoscale.go

@@ -74,7 +74,7 @@ const (
 //    * increase_events -- number of times the replica count was increased.
 //    * decrease_events -- number of times the replica count was decreased.
 //    * cooldown_drops  -- number of times an event was dropped because of a cooldown timer.
-//    * errors          -- number of errors communicating with the Kubernetes API.
+//    * errors          -- number of errors encountered, typically related to communicating with the Kubernetes API.
 //
 type K8sAutoscaleNode struct {
 	chainnode
@@ -83,7 +83,7 @@ type K8sAutoscaleNode struct {
 	Namespace string
 
 	// Kind is the type of resources to autoscale.
-	// Currently only "deployments" and "replicationcontrollers" are supported.
+	// Currently only "deployments", "replicasets" and "replicationcontrollers" are supported.
 	// Default: "deployments"
 	Kind string
 
@@ -155,8 +155,8 @@ func (n *K8sAutoscaleNode) validate() error {
 		(n.ResourceNameTag == "" && n.ResourceName == "") {
 		return fmt.Errorf("must specify exactly one of ResourceName or ResourceNameTag")
 	}
-	if n.Kind != client.DeploymentsKind && n.Kind != client.ReplicationControllerKind {
-		return fmt.Errorf("invalid Kind, must be 'deployments' or 'replicationcontrollers', got %s", n.Kind)
+	if n.Kind != client.DeploymentsKind && n.Kind != client.ReplicationControllerKind && n.Kind != client.ReplicaSetsKind {
+		return fmt.Errorf("invalid Kind, must be 'deployments', 'replicasets' or 'replicationcontrollers', got %s", n.Kind)
 	}
 	if n.Min < 1 {
 		return fmt.Errorf("min must be >= 1, got %d", n.Min)

services/k8s/client/types.go

@@ -10,6 +10,7 @@ const (
 
 const (
 	DeploymentsKind           = "deployments"
+	ReplicaSetsKind           = "replicasets"
 	ReplicationControllerKind = "replicationcontroller"
 )