CSPL-4360 Add Splunk restart

Author: kasiakoziol

docs/IndexIngestionSeparation.md

@@ -1,3 +1,9 @@
+---
+title: Index and Ingestion Separation
+parent: Deploy & Configure
+nav_order: 6
+---
+
 # Background
 
 Separation between ingestion and indexing services within Splunk Operator for Kubernetes enables the operator to independently manage the ingestion service while maintaining seamless integration with the indexing service.
@@ -10,7 +16,7 @@ This separation enables:
 # Important Note
 
 > [!WARNING]
-> **As of now, only brand new deployments are supported for Index and Ingestion Separation. No migration path is implemented, described or tested for existing deployments to move from a standard model to Index & Ingestion separation model.**
+> **For customers deploying SmartBus on CMP, the Splunk Operator for Kubernetes (SOK) manages the configuration and lifecycle of the ingestor tier. The following SOK guide provides implementation details for setting up ingestion separation and integrating with existing indexers. This reference is primarily intended for CMP users leveraging SOK-managed ingestors.**
 
 # Document Variables
 
@@ -38,7 +44,7 @@ SQS message bus inputs can be found in the table below.
 | endpoint   | string | [Optional, if not provided formed based on region] AWS SQS Service endpoint
 | dlq   | string | [Required] Name of the dead letter queue |
 
-Change of any of the bus inputs triggers the restart of Splunk so that appropriate .conf files are correctly refreshed and consumed.
+**First provisioning or update of any of the bus inputs requires Ingestor Cluster and Indexer Cluster Splunkd restart, but this restart is implemented automatically and done by SOK.**
 
 ## Example
 ```
@@ -425,6 +431,14 @@ In the following example, the dashboard presents ingestion and indexing data in
 
 - [kube-prometheus-stack](https://github.com/prometheus-community/helm-charts/tree/main/charts/kube-prometheus-stack)
 
+# App Installation for Ingestor Cluster Instances
+
+Application installation is supported for Ingestor Cluster instances. However, as of now, applications are installed using local scope and if any application requires Splunk restart, there is no automated way to detect it and trigger automatically via Splunk Operator. 
+
+Therefore, to be able to enforce Splunk restart for each of the Ingestor Cluster pods, it is recommended to add/update IngestorCluster CR annotations/labels and apply the new configuration which will trigger the rolling restart of Splunk pods for Ingestor Cluster. 
+
+We are under the investigation on how to make it fully automated. What is more, ideally, update of annotations and labels should not trigger pod restart at all and we are investigating on how to fix this behaviour eventually.
+
 # Example
 
 1. Install CRDs and Splunk Operator for Kubernetes.

pkg/splunk/enterprise/indexercluster.go

@@ -305,6 +305,15 @@ func ApplyIndexerClusterManager(ctx context.Context, client splcommon.Controller
 				}
 
 				cr.Status.Bus = &bus.Spec
+
+				for i := int32(0); i < cr.Spec.Replicas; i++ {
+					idxcClient := mgr.getClient(ctx, i)
+					err = idxcClient.RestartSplunk()
+					if err != nil {
+						return result, err
+					}
+					scopedLog.Info("Restarted splunk", "indexer", i)
+				}
 			}
 		}
 
@@ -627,6 +636,15 @@ func ApplyIndexerCluster(ctx context.Context, client splcommon.ControllerClient,
 				}
 
 				cr.Status.Bus = &bus.Spec
+
+				for i := int32(0); i < cr.Spec.Replicas; i++ {
+					idxcClient := mgr.getClient(ctx, i)
+					err = idxcClient.RestartSplunk()
+					if err != nil {
+						return result, err
+					}
+					scopedLog.Info("Restarted splunk", "indexer", i)
+				}
 			}
 		}
 

pkg/splunk/enterprise/ingestorcluster.go

@@ -260,7 +260,7 @@ func ApplyIngestorCluster(ctx context.Context, client client.Client, cr *enterpr
 
 		// If bus is updated
 		if !reflect.DeepEqual(cr.Status.Bus, bus.Spec) || !reflect.DeepEqual(cr.Status.LargeMessageStore, lms.Spec) {
-			mgr := newIngestorClusterPodManager(scopedLog, cr, namespaceScopedSecret, splclient.NewSplunkClient)
+			mgr := newIngestorClusterPodManager(scopedLog, cr, namespaceScopedSecret, splclient.NewSplunkClient, client)
 			err = mgr.handlePushBusChange(ctx, cr, busCopy, lmsCopy, client)
 			if err != nil {
 				eventPublisher.Warning(ctx, "ApplyIngestorCluster", fmt.Sprintf("Failed to update conf file for Bus/Pipeline config change after pod creation: %s", err.Error()))
@@ -269,6 +269,15 @@ func ApplyIngestorCluster(ctx context.Context, client client.Client, cr *enterpr
 			}
 
 			cr.Status.Bus = &bus.Spec
+
+			for i := int32(0); i < cr.Spec.Replicas; i++ {
+				ingClient := mgr.getClient(ctx, i)
+				err = ingClient.RestartSplunk()
+				if err != nil {
+					return result, err
+				}
+				scopedLog.Info("Restarted splunk", "ingestor", i)
+			}
 		}
 
 		// Upgrade fron automated MC to MC CRD
@@ -311,6 +320,27 @@ func ApplyIngestorCluster(ctx context.Context, client client.Client, cr *enterpr
 	return result, nil
 }
 
+// getClient for ingestorClusterPodManager returns a SplunkClient for the member n
+func (mgr *ingestorClusterPodManager) getClient(ctx context.Context, n int32) *splclient.SplunkClient {
+	reqLogger := log.FromContext(ctx)
+	scopedLog := reqLogger.WithName("ingestorClusterPodManager.getClient").WithValues("name", mgr.cr.GetName(), "namespace", mgr.cr.GetNamespace())
+
+	// Get Pod Name
+	memberName := GetSplunkStatefulsetPodName(SplunkIngestor, mgr.cr.GetName(), n)
+
+	// Get Fully Qualified Domain Name
+	fqdnName := splcommon.GetServiceFQDN(mgr.cr.GetNamespace(),
+		fmt.Sprintf("%s.%s", memberName, GetSplunkServiceName(SplunkIngestor, mgr.cr.GetName(), true)))
+
+	// Retrieve admin password from Pod
+	adminPwd, err := splutil.GetSpecificSecretTokenFromPod(ctx, mgr.c, memberName, mgr.cr.GetNamespace(), "password")
+	if err != nil {
+		scopedLog.Error(err, "Couldn't retrieve the admin password from pod")
+	}
+
+	return mgr.newSplunkClient(fmt.Sprintf("https://%s:8089", fqdnName), "admin", adminPwd)
+}
+
 // validateIngestorClusterSpec checks validity and makes default updates to a IngestorClusterSpec and returns error if something is wrong
 func validateIngestorClusterSpec(ctx context.Context, c splcommon.ControllerClient, cr *enterpriseApi.IngestorCluster) error {
 	// We cannot have 0 replicas in IngestorCluster spec since this refers to number of ingestion pods in an ingestor cluster
@@ -426,19 +456,21 @@ func getChangedBusFieldsForIngestor(bus *enterpriseApi.Bus, lms *enterpriseApi.L
 }
 
 type ingestorClusterPodManager struct {
+	c               splcommon.ControllerClient
 	log             logr.Logger
 	cr              *enterpriseApi.IngestorCluster
 	secrets         *corev1.Secret
 	newSplunkClient func(managementURI, username, password string) *splclient.SplunkClient
 }
 
 // newIngestorClusterPodManager function to create pod manager this is added to write unit test case
-var newIngestorClusterPodManager = func(log logr.Logger, cr *enterpriseApi.IngestorCluster, secret *corev1.Secret, newSplunkClient NewSplunkClientFunc) ingestorClusterPodManager {
+var newIngestorClusterPodManager = func(log logr.Logger, cr *enterpriseApi.IngestorCluster, secret *corev1.Secret, newSplunkClient NewSplunkClientFunc, c splcommon.ControllerClient) ingestorClusterPodManager {
 	return ingestorClusterPodManager{
 		log:             log,
 		cr:              cr,
 		secrets:         secret,
 		newSplunkClient: newSplunkClient,
+		c:               c,
 	}
 }
 

pkg/splunk/enterprise/ingestorcluster_test.go

@@ -25,15 +25,14 @@ import (
 	"github.com/go-logr/logr"
 	enterpriseApi "github.com/splunk/splunk-operator/api/v4"
 	splclient "github.com/splunk/splunk-operator/pkg/splunk/client"
+	splcommon "github.com/splunk/splunk-operator/pkg/splunk/common"
 	spltest "github.com/splunk/splunk-operator/pkg/splunk/test"
 	splutil "github.com/splunk/splunk-operator/pkg/splunk/util"
 	"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"
-	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 func init() {
@@ -56,11 +55,7 @@ func TestApplyIngestorCluster(t *testing.T) {
 
 	ctx := context.TODO()
 
-	scheme := runtime.NewScheme()
-	_ = enterpriseApi.AddToScheme(scheme)
-	_ = corev1.AddToScheme(scheme)
-	_ = appsv1.AddToScheme(scheme)
-	c := fake.NewClientBuilder().WithScheme(scheme).Build()
+	c := spltest.NewMockClient()
 
 	// Object definitions
 	provider := "sqs_smartbus"
@@ -273,8 +268,9 @@ func TestApplyIngestorCluster(t *testing.T) {
 	// outputs.conf
 	origNew := newIngestorClusterPodManager
 	mockHTTPClient := &spltest.MockHTTPClient{}
-	newIngestorClusterPodManager = func(l logr.Logger, cr *enterpriseApi.IngestorCluster, secret *corev1.Secret, _ NewSplunkClientFunc) ingestorClusterPodManager {
+	newIngestorClusterPodManager = func(l logr.Logger, cr *enterpriseApi.IngestorCluster, secret *corev1.Secret, _ NewSplunkClientFunc, c splcommon.ControllerClient) ingestorClusterPodManager {
 		return ingestorClusterPodManager{
+			c:   c,
 			log: l, cr: cr, secrets: secret,
 			newSplunkClient: func(uri, user, pass string) *splclient.SplunkClient {
 				return &splclient.SplunkClient{ManagementURI: uri, Username: user, Password: pass, Client: mockHTTPClient}

pkg/splunk/enterprise/util_test.go

@@ -2624,6 +2624,9 @@ func TestUpdateCRStatus(t *testing.T) {
 		WithStatusSubresource(&enterpriseApi.Standalone{}).
 		WithStatusSubresource(&enterpriseApi.MonitoringConsole{}).
 		WithStatusSubresource(&enterpriseApi.IndexerCluster{}).
+		WithStatusSubresource(&enterpriseApi.Bus{}).
+		WithStatusSubresource(&enterpriseApi.LargeMessageStore{}).
+		WithStatusSubresource(&enterpriseApi.IngestorCluster{}).
 		WithStatusSubresource(&enterpriseApi.SearchHeadCluster{})
 	c := builder.Build()
 	ctx := context.TODO()
@@ -3304,6 +3307,8 @@ func TestGetCurrentImage(t *testing.T) {
 		WithStatusSubresource(&enterpriseApi.MonitoringConsole{}).
 		WithStatusSubresource(&enterpriseApi.IndexerCluster{}).
 		WithStatusSubresource(&enterpriseApi.SearchHeadCluster{}).
+		WithStatusSubresource(&enterpriseApi.Bus{}).
+		WithStatusSubresource(&enterpriseApi.LargeMessageStore{}).
 		WithStatusSubresource(&enterpriseApi.IngestorCluster{})
 	client := builder.Build()
 	client.Create(ctx, &current)

test/index_and_ingestion_separation/index_and_ingestion_separation_test.go

@@ -433,14 +433,6 @@ var _ = Describe("indingsep test", func() {
 			err = deployment.UpdateCR(ctx, bus)
 			Expect(err).To(Succeed(), "Unable to deploy Bus with updated CR")
 
-			// Ensure that Ingestor Cluster has not been restarted
-			testcaseEnvInst.Log.Info("Ensure that Ingestor Cluster has not been restarted")
-			testenv.IngestorReady(ctx, deployment, testcaseEnvInst)
-
-			// Ensure that Indexer Cluster has not been restarted
-			testcaseEnvInst.Log.Info("Ensure that Indexer Cluster has not been restarted")
-			testenv.SingleSiteIndexersReady(ctx, deployment, testcaseEnvInst)
-
 			// Get instance of current Ingestor Cluster CR with latest config
 			testcaseEnvInst.Log.Info("Get instance of current Ingestor Cluster CR with latest config")
 			ingest := &enterpriseApi.IngestorCluster{}