Added consistent hashing strategy

apis/v1alpha1/opentelemetrycollector_types.go

@@ -116,7 +116,14 @@ type OpenTelemetryCollectorSpec struct {
 
 // OpenTelemetryTargetAllocator defines the configurations for the Prometheus target allocator.
 type OpenTelemetryTargetAllocator struct {
+	// Replicas is the number of pod instances for the underlying TargetAllocator, this should only be set to a value
+	// other than 1 if a strategy that allows for high availability is chosen. Currently, the only allocation strategy
+	// that can be run in a high availability mode is consistent-hashing.
+	// +optional
+	Replicas *int32 `json:"replicas,omitempty"`
 	// AllocationStrategy determines which strategy the target allocator should use for allocation
+	// The current options are least-weighted and consistent-hashing. The default option is least-weighted
+	// +optional
 	AllocationStrategy string `json:"allocationStrategy,omitempty"`
 	// ServiceAccount indicates the name of an existing service account to use with this instance.
 	// +optional

apis/v1alpha1/opentelemetrycollector_webhook.go

@@ -56,12 +56,15 @@ func (r *OpenTelemetryCollector) Default() {
 		r.Labels["app.kubernetes.io/managed-by"] = "opentelemetry-operator"
 	}
 
+	// We can default to one because dependent objects Deployment and HorizontalPodAutoScaler
+	// default to 1 as well.
+	one := int32(1)
 	if r.Spec.Replicas == nil {
-		// We can default to one because dependent objects Deployment and HorizontalPodAutoScaler
-		// default to 1 as well.
-		one := int32(1)
 		r.Spec.Replicas = &one
 	}
+	if r.Spec.TargetAllocator.Enabled && r.Spec.TargetAllocator.Replicas == nil {
+		r.Spec.TargetAllocator.Replicas = &one
+	}
 }
 
 // +kubebuilder:webhook:verbs=create;update,path=/validate-opentelemetry-io-v1alpha1-opentelemetrycollector,mutating=false,failurePolicy=fail,groups=opentelemetry.io,resources=opentelemetrycollectors,versions=v1alpha1,name=vopentelemetrycollectorcreateupdate.kb.io,sideEffects=none,admissionReviewVersions=v1

bundle/manifests/opentelemetry.io_opentelemetrycollectors.yaml

@@ -809,7 +809,9 @@ spec:
                 properties:
                   allocationStrategy:
                     description: AllocationStrategy determines which strategy the
-                      target allocator should use for allocation
+                      target allocator should use for allocation The current options
+                      are least-weighted and consistent-hashing. The default option
+                      is least-weighted
                     type: string
                   enabled:
                     description: Enabled indicates whether to use a target allocation
@@ -831,6 +833,14 @@ spec:
                           custom resources as targets or not.
                         type: boolean
                     type: object
+                  replicas:
+                    description: Replicas is the number of pod instances for the underlying
+                      TargetAllocator, this should only be set to a value other than
+                      1 if a strategy that allows for high availability is chosen.
+                      Currently, the only allocation strategy that can be run in a
+                      high availability mode is consistent-hashing.
+                    format: int32
+                    type: integer
                   serviceAccount:
                     description: ServiceAccount indicates the name of an existing
                       service account to use with this instance.

cmd/otel-allocator/allocation/consistent_hashing.go

@@ -2,13 +2,15 @@ package allocation
 
 import (
 	"fmt"
+	"net/url"
+	"sync"
+
 	"github.com/buraksezer/consistent"
 	"github.com/cespare/xxhash/v2"
 	"github.com/go-logr/logr"
-	"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/diff"
 	"github.com/prometheus/client_golang/prometheus"
-	"net/url"
-	"sync"
+
+	"github.com/open-telemetry/opentelemetry-operator/cmd/otel-allocator/diff"
 )
 
 var _ Allocator = &consistentHashingAllocator{}

cmd/otel-allocator/allocation/consistent_hashing_test.go

@@ -1,8 +1,9 @@
 package allocation
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestCanSetSingleTarget(t *testing.T) {

cmd/otel-allocator/allocation/http_test.go

@@ -159,15 +159,15 @@ func TestGetAllTargetsByCollectorAndJob(t *testing.T) {
 								"test-label": "test-value",
 								"foo":        "bar",
 							},
-							TargetURL: "test-url",
+							TargetURL:     "test-url",
 							CollectorName: "test-collector",
 						},
 						TargetItem{
 							JobName: "test-job",
 							Label: model.LabelSet{
 								"test-label": "test-value",
 							},
-							TargetURL: "test-url",
+							TargetURL:     "test-url",
 							CollectorName: "test-collector",
 						},
 					},

cmd/otel-allocator/allocation/strategy.go

@@ -3,9 +3,9 @@ package allocation
 import (
 	"errors"
 	"fmt"
-	"github.com/buraksezer/consistent"
 	"net/url"
 
+	"github.com/buraksezer/consistent"
 	"github.com/go-logr/logr"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"

config/crd/bases/opentelemetry.io_opentelemetrycollectors.yaml

@@ -807,7 +807,9 @@ spec:
                 properties:
                   allocationStrategy:
                     description: AllocationStrategy determines which strategy the
-                      target allocator should use for allocation
+                      target allocator should use for allocation The current options
+                      are least-weighted and consistent-hashing. The default option
+                      is least-weighted
                     type: string
                   enabled:
                     description: Enabled indicates whether to use a target allocation
@@ -829,6 +831,14 @@ spec:
                           custom resources as targets or not.
                         type: boolean
                     type: object
+                  replicas:
+                    description: Replicas is the number of pod instances for the underlying
+                      TargetAllocator, this should only be set to a value other than
+                      1 if a strategy that allows for high availability is chosen.
+                      Currently, the only allocation strategy that can be run in a
+                      high availability mode is consistent-hashing.
+                    format: int32
+                    type: integer
                   serviceAccount:
                     description: ServiceAccount indicates the name of an existing
                       service account to use with this instance.

docs/api.md

@@ -3116,7 +3116,7 @@ TargetAllocator indicates a value which determines whether to spawn a target all
         <td><b>allocationStrategy</b></td>
         <td>string</td>
         <td>
-          AllocationStrategy determines which strategy the target allocator should use for allocation<br/>
+          AllocationStrategy determines which strategy the target allocator should use for allocation The current options are least-weighted and consistent-hashing. The default option is least-weighted<br/>
         </td>
         <td>false</td>
       </tr><tr>
@@ -3140,6 +3140,15 @@ TargetAllocator indicates a value which determines whether to spawn a target all
           PrometheusCR defines the configuration for the retrieval of PrometheusOperator CRDs ( servicemonitor.monitoring.coreos.com/v1 and podmonitor.monitoring.coreos.com/v1 )  retrieval. All CR instances which the ServiceAccount has access to will be retrieved. This includes other namespaces.<br/>
         </td>
         <td>false</td>
+      </tr><tr>
+        <td><b>replicas</b></td>
+        <td>integer</td>
+        <td>
+          Replicas is the number of pod instances for the underlying TargetAllocator, this should only be set to a value other than 1 if a strategy that allows for high availability is chosen. Currently, the only allocation strategy that can be run in a high availability mode is consistent-hashing.<br/>
+          <br/>
+            <i>Format</i>: int32<br/>
+        </td>
+        <td>false</td>
       </tr><tr>
         <td><b>serviceAccount</b></td>
         <td>string</td>

pkg/collector/reconcile/deployment_test.go

@@ -127,6 +127,7 @@ func TestExpectedDeployments(t *testing.T) {
 
 	t.Run("should not update target allocator deployment replicas when collector max replicas is set", func(t *testing.T) {
 		replicas, maxReplicas := int32(2), int32(10)
+		oneReplica := int32(1)
 		param := Params{
 			Client: k8sClient,
 			Instance: v1alpha1.OpenTelemetryCollector{
@@ -144,7 +145,8 @@ func TestExpectedDeployments(t *testing.T) {
 					Replicas:    &replicas,
 					Mode:        v1alpha1.ModeStatefulSet,
 					TargetAllocator: v1alpha1.OpenTelemetryTargetAllocator{
-						Enabled: true,
+						Enabled:  true,
+						Replicas: &oneReplica,
 					},
 					Config: `
 				receivers:
@@ -177,6 +179,61 @@ func TestExpectedDeployments(t *testing.T) {
 		assert.Equal(t, *allocator.Spec.Replicas, int32(1))
 	})
 
+	t.Run("should update target allocator deployment replicas when changed", func(t *testing.T) {
+		initialReplicas, nextReplicas := int32(1), int32(2)
+		param := Params{
+			Client: k8sClient,
+			Instance: v1alpha1.OpenTelemetryCollector{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "opentelemetry.io",
+					APIVersion: "v1",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test",
+					Namespace: "default",
+					UID:       instanceUID,
+				},
+				Spec: v1alpha1.OpenTelemetryCollectorSpec{
+					Replicas: &initialReplicas,
+					Mode:     v1alpha1.ModeStatefulSet,
+					TargetAllocator: v1alpha1.OpenTelemetryTargetAllocator{
+						Enabled:  true,
+						Replicas: &initialReplicas,
+					},
+					Config: `
+				receivers:
+				jaeger:
+					protocols:
+					grpc:
+				processors:
+
+				exporters:
+				logging:
+
+				service:
+				pipelines:
+					traces:
+					receivers: [jaeger]
+					processors: []
+					exporters: [logging]
+
+			`,
+				},
+			},
+			Scheme: testScheme,
+			Log:    logger,
+		}
+		expected := []v1.Deployment{}
+		allocator := targetallocator.Deployment(param.Config, param.Log, param.Instance)
+		expected = append(expected, allocator)
+
+		assert.Len(t, expected, 1)
+		assert.Equal(t, *allocator.Spec.Replicas, int32(1))
+		param.Instance.Spec.TargetAllocator.Replicas = &nextReplicas
+		finalAllocator := targetallocator.Deployment(param.Config, param.Log, param.Instance)
+		assert.Equal(t, *finalAllocator.Spec.Replicas, int32(2))
+	})
+
 	t.Run("should update deployment", func(t *testing.T) {
 		createObjectIfNotExists(t, "test-collector", &expectedDeploy)
 		err := expectedDeployments(context.Background(), param, []v1.Deployment{expectedDeploy})

pkg/targetallocator/deployment.go

@@ -30,16 +30,14 @@ func Deployment(cfg config.Config, logger logr.Logger, otelcol v1alpha1.OpenTele
 	labels := Labels(otelcol)
 	labels["app.kubernetes.io/name"] = naming.TargetAllocator(otelcol)
 
-	var replicas int32 = 1
-
 	return appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      naming.TargetAllocator(otelcol),
 			Namespace: otelcol.Namespace,
 			Labels:    labels,
 		},
 		Spec: appsv1.DeploymentSpec{
-			Replicas: &replicas,
+			Replicas: otelcol.Spec.TargetAllocator.Replicas,
 			Selector: &metav1.LabelSelector{
 				MatchLabels: labels,
 			},