Responded to code review commits

Also updated to take account of pipelines

pkg/controller/logstash/logstash_controller_test.go

@@ -9,15 +9,19 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
 
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	commonv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/common/v1"
 	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/comparison"
@@ -38,12 +42,13 @@ func newReconcileLogstash(objs ...runtime.Object) *ReconcileLogstash {
 func TestReconcileLogstash_Reconcile(t *testing.T) {
 	defaultLabels := (&logstashv1alpha1.Logstash{ObjectMeta: metav1.ObjectMeta{Name: "testLogstash"}}).GetIdentityLabels()
 	tests := []struct {
-		name     string
-		objs     []runtime.Object
-		request  reconcile.Request
-		want     reconcile.Result
-		expected logstashv1alpha1.Logstash
-		wantErr  bool
+		name            string
+		objs            []runtime.Object
+		request         reconcile.Request
+		want            reconcile.Result
+		expected        logstashv1alpha1.Logstash
+		expectedObjects expectedObjects
+		wantErr         bool
 	}{
 		{
 			name: "valid unmanaged Logstash does not increment observedGeneration",
@@ -88,7 +93,8 @@ func TestReconcileLogstash_Reconcile(t *testing.T) {
 					ObservedGeneration: 1,
 				},
 			},
-			wantErr: false,
+			expectedObjects: []expectedObject{},
+			wantErr:         false,
 		},
 		{
 			name: "too long name fails validation, and updates observedGeneration",
@@ -121,10 +127,93 @@ func TestReconcileLogstash_Reconcile(t *testing.T) {
 					ObservedGeneration: 2,
 				},
 			},
-			wantErr: true,
+			expectedObjects: []expectedObject{},
+			wantErr:         true,
+		},
+		{
+			name: "Logstash with ready StatefulSet and Pod updates status and creates secrets and service",
+			objs: []runtime.Object{
+				&logstashv1alpha1.Logstash{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:       "testLogstash",
+						Namespace:  "test",
+						Generation: 2,
+					},
+					Spec: logstashv1alpha1.LogstashSpec{
+						Version: "8.6.1",
+						Count:   1,
+					},
+					Status: logstashv1alpha1.LogstashStatus{
+						ObservedGeneration: 1,
+					},
+				},
+				&appsv1.StatefulSet{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "testLogstash-ls",
+						Namespace: "test",
+						Labels:    addLabel(defaultLabels, hash.TemplateHashLabelName, "3145706383"),
+					},
+					Status: appsv1.StatefulSetStatus{
+						AvailableReplicas: 1,
+						Replicas:          1,
+						ReadyReplicas:     1,
+					},
+				},
+				&corev1.Pod{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:       "testLogstash-ls",
+						Namespace:  "test",
+						Generation: 2,
+						Labels:     map[string]string{NameLabelName: "testLogstash", VersionLabelName: "8.6.1"},
+					},
+					Status: corev1.PodStatus{
+						Phase: corev1.PodRunning,
+					},
+				},
+			},
+			request: reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Namespace: "test",
+					Name:      "testLogstash",
+				},
+			},
+			want: reconcile.Result{},
+			expectedObjects: []expectedObject{
+				{
+					t:    &corev1.Service{},
+					name: types.NamespacedName{Namespace: "test", Name: "testLogstash-ls-api"},
+				},
+				{
+					t:    &corev1.Secret{},
+					name: types.NamespacedName{Namespace: "test", Name: "testLogstash-ls-config"},
+				},
+				{
+					t:    &corev1.Secret{},
+					name: types.NamespacedName{Namespace: "test", Name: "testLogstash-ls-pipeline"},
+				},
+			},
+
+			expected: logstashv1alpha1.Logstash{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "testLogstash",
+					Namespace:  "test",
+					Generation: 2,
+				},
+				Spec: logstashv1alpha1.LogstashSpec{
+					Version: "8.6.1",
+					Count:   1,
+				},
+				Status: logstashv1alpha1.LogstashStatus{
+					Version:            "8.6.1",
+					ExpectedNodes:      1,
+					AvailableNodes:     1,
+					ObservedGeneration: 2,
+				},
+			},
+			wantErr: false,
 		},
 		{
-			name: "Logstash with ready stateful set and pod updates status properly",
+			name: "Logstash with a custom service creates secrets and service",
 			objs: []runtime.Object{
 				&logstashv1alpha1.Logstash{
 					ObjectMeta: metav1.ObjectMeta{
@@ -135,6 +224,16 @@ func TestReconcileLogstash_Reconcile(t *testing.T) {
 					Spec: logstashv1alpha1.LogstashSpec{
 						Version: "8.6.1",
 						Count:   1,
+						Services: []logstashv1alpha1.LogstashService{{
+							Name: "test",
+							Service: commonv1.ServiceTemplate{
+								Spec: corev1.ServiceSpec{
+									Ports: []corev1.ServicePort{
+										{Protocol: "TCP", Port: 9500},
+									},
+								},
+							},
+						}},
 					},
 					Status: logstashv1alpha1.LogstashStatus{
 						ObservedGeneration: 1,
@@ -171,6 +270,25 @@ func TestReconcileLogstash_Reconcile(t *testing.T) {
 				},
 			},
 			want: reconcile.Result{},
+			expectedObjects: []expectedObject{
+				{
+					t:    &corev1.Service{},
+					name: types.NamespacedName{Namespace: "test", Name: "testLogstash-ls-api"},
+				},
+				{
+					t:    &corev1.Service{},
+					name: types.NamespacedName{Namespace: "test", Name: "testLogstash-ls-test"},
+				},
+				{
+					t:    &corev1.Secret{},
+					name: types.NamespacedName{Namespace: "test", Name: "testLogstash-ls-config"},
+				},
+				{
+					t:    &corev1.Secret{},
+					name: types.NamespacedName{Namespace: "test", Name: "testLogstash-ls-pipeline"},
+				},
+			},
+
 			expected: logstashv1alpha1.Logstash{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:       "testLogstash",
@@ -180,6 +298,16 @@ func TestReconcileLogstash_Reconcile(t *testing.T) {
 				Spec: logstashv1alpha1.LogstashSpec{
 					Version: "8.6.1",
 					Count:   1,
+					Services: []logstashv1alpha1.LogstashService{{
+						Name: "test",
+						Service: commonv1.ServiceTemplate{
+							Spec: corev1.ServiceSpec{
+								Ports: []corev1.ServicePort{
+									{Protocol: "TCP", Port: 9500},
+								},
+							},
+						},
+					}},
 				},
 				Status: logstashv1alpha1.LogstashStatus{
 					Version:            "8.6.1",
@@ -208,7 +336,7 @@ func TestReconcileLogstash_Reconcile(t *testing.T) {
 				t.Error(err)
 				return
 			}
-
+			tt.expectedObjects.assertExist(t, r.Client)
 			comparison.AssertEqual(t, &Logstash, &tt.expected)
 		})
 	}
@@ -222,3 +350,18 @@ func addLabel(labels map[string]string, key, value string) map[string]string {
 	newLabels[key] = value
 	return newLabels
 }
+
+type expectedObject struct {
+	t    client.Object
+	name types.NamespacedName
+}
+
+type expectedObjects []expectedObject
+
+func (e expectedObjects) assertExist(t *testing.T, k8s client.Client) {
+	t.Helper()
+	for _, o := range e {
+		obj := o.t.DeepCopyObject().(client.Object) //nolint:forcetypeassert
+		assert.NoError(t, k8s.Get(context.Background(), o.name, obj), "Expected object not found: %s", o.name)
+	}
+}

pkg/controller/logstash/pod.go

@@ -17,7 +17,6 @@ import (
 	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/container"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/defaults"
-	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/pod"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/tracing"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/volume"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash/network"
@@ -123,7 +122,7 @@ func readinessProbe(logstash logstashv1alpha1.Logstash) corev1.Probe {
 	var scheme = corev1.URISchemeHTTP
 	var port = network.HTTPPort
 	for _, service := range logstash.Spec.Services {
-		if service.Name == "api" {
+		if service.Name == LogstashAPIServiceName {
 			port = int(service.Service.Spec.Ports[0].Port)
 		}
 	}
@@ -142,9 +141,4 @@ func readinessProbe(logstash logstashv1alpha1.Logstash) corev1.Probe {
 		},
 	}
 	return probe
-}
-
-// GetLogstashContainer returns the Logstash container from the given podSpec.
-func GetLogstashContainer(podSpec corev1.PodSpec) *corev1.Container {
-	return pod.ContainerByName(podSpec, logstashv1alpha1.LogstashContainerName)
-}
+}

pkg/controller/logstash/pod_test.go

@@ -12,15 +12,15 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	corev1 "k8s.io/api/core/v1"
-
-	commonv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/common/v1"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	commonv1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/common/v1"
 	logstashv1alpha1 "github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/container"
+	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/common/pod"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/utils/k8s"
 )
 
@@ -41,11 +41,11 @@ func TestNewPodTemplateSpec(t *testing.T) {
 				assert.Equal(t, false, *pod.Spec.AutomountServiceAccountToken)
 				assert.Len(t, pod.Spec.Containers, 1)
 				assert.Len(t, pod.Spec.InitContainers, 0)
-				assert.Len(t, pod.Spec.Volumes, 1)
+				assert.Len(t, pod.Spec.Volumes, 2)
 				assert.NotEmpty(t, pod.Annotations[ConfigHashAnnotationName])
 				logstashContainer := GetLogstashContainer(pod.Spec)
 				require.NotNil(t, logstashContainer)
-				assert.Equal(t, 1, len(logstashContainer.VolumeMounts))
+				assert.Equal(t, 2, len(logstashContainer.VolumeMounts))
 				assert.Equal(t, container.ImageRepository(container.LogstashImage, "8.6.1"), logstashContainer.Image)
 				assert.NotNil(t, logstashContainer.ReadinessProbe)
 				assert.NotEmpty(t, logstashContainer.Ports)
@@ -171,7 +171,7 @@ func TestNewPodTemplateSpec(t *testing.T) {
 				Spec: logstashv1alpha1.LogstashSpec{
 					Version: "8.6.1",
 					Services: []logstashv1alpha1.LogstashService{{
-						Name: "api",
+						Name: LogstashAPIServiceName,
 						Service: commonv1.ServiceTemplate{
 							Spec: corev1.ServiceSpec{
 								Ports: []corev1.ServicePort{
@@ -201,7 +201,7 @@ func TestNewPodTemplateSpec(t *testing.T) {
 					Version: "8.6.1",
 					Services: []logstashv1alpha1.LogstashService{
 						{
-							Name: "api",
+							Name: LogstashAPIServiceName,
 							Service: commonv1.ServiceTemplate{
 								Spec: corev1.ServiceSpec{
 									Ports: []corev1.ServicePort{
@@ -260,8 +260,8 @@ func TestNewPodTemplateSpec(t *testing.T) {
 				},
 			}},
 			assertions: func(pod corev1.PodTemplateSpec) {
-				assert.Len(t, pod.Spec.Volumes, 2)
-				assert.Len(t, GetLogstashContainer(pod.Spec).VolumeMounts, 2)
+				assert.Len(t, pod.Spec.Volumes, 3)
+				assert.Len(t, GetLogstashContainer(pod.Spec).VolumeMounts, 3)
 			},
 		},
 	}
@@ -277,4 +277,9 @@ func TestNewPodTemplateSpec(t *testing.T) {
 			tt.assertions(got)
 		})
 	}
-}
+}
+
+// GetLogstashContainer returns the Logstash container from the given podSpec.
+func GetLogstashContainer(podSpec corev1.PodSpec) *corev1.Container {
+	return pod.ContainerByName(podSpec, logstashv1alpha1.LogstashContainerName)
+}

pkg/controller/logstash/service.go

@@ -14,6 +14,10 @@ import (
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash/network"
 )
 
+const (
+	LogstashAPIServiceName = "api"
+)
+
 // reconcileServices reconcile Services defined in spec
 //
 // When a service is defined that matches the API service name, then that service is used to define
@@ -84,7 +88,7 @@ func newAPIService(logstash logstashv1alpha1.Logstash) *corev1.Service {
 	labels := NewLabels(logstash)
 	ports := []corev1.ServicePort{
 		{
-			Name:     "api",
+			Name:     LogstashAPIServiceName,
 			Protocol: corev1.ProtocolTCP,
 			Port:     network.HTTPPort,
 		},

pkg/controller/logstash/services_test.go → pkg/controller/logstash/service_test.go

@@ -73,11 +73,11 @@ func TestReconcileServices(t *testing.T) {
 				},
 				Spec: logstashv1alpha1.LogstashSpec{
 					Services: []logstashv1alpha1.LogstashService{{
-						Name: "api",
+						Name: LogstashAPIServiceName,
 						Service: commonv1.ServiceTemplate{
 							Spec: corev1.ServiceSpec{
 								Ports: []corev1.ServicePort{
-									{Name: "api", Protocol: "TCP", Port: 9200},
+									{Name: LogstashAPIServiceName, Protocol: "TCP", Port: 9200},
 								},
 							},
 						},

test/e2e/test/logstash/http_client.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/elastic/cloud-on-k8s/v2/pkg/apis/logstash/v1alpha1"
 	"github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash/network"
+	ls "github.com/elastic/cloud-on-k8s/v2/pkg/controller/logstash"
 	"github.com/elastic/cloud-on-k8s/v2/test/e2e/test"
 )
 
@@ -35,7 +36,7 @@ func DoRequest(client *http.Client, logstash v1alpha1.Logstash, method, path str
 	var scheme = "http"
 	var port = network.HTTPPort
 	for _, service := range logstash.Spec.Services {
-		if service.Name == "api" {
+		if service.Name == ls.LogstashAPIServiceName {
 			port = int(service.Service.Spec.Ports[0].Port)
 		}
 	}