Discover Index ID

Discover Index ID by spawning a pod with the rag-content container
image. By inspecting the value stored in INDEX_NAME env variable we can
conclude what Index ID value should be utilized [1].

This is only temporarly solution until the IndexID gets removed from
required fields in OLSConfig.

[1]
https://github.com/openstack-lightspeed/rag-content/blob/5e8339870c3600120bf6080e68a6ffb14a115d61/Containerfile#L53C5-L53C15

Author: lpiwowar

bindata/rbac/rbac.yaml

@@ -113,6 +113,12 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods/log
+  verbs:
+  - get
 - apiGroups:
   - ""
   resources:

config/rbac/role.yaml

@@ -70,6 +70,12 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods/log
+  verbs:
+  - get
 - apiGroups:
   - ""
   resources:

controllers/lightspeed/openstacklightspeed_controller.go

@@ -58,6 +58,9 @@ func (r *OpenStackLightspeedReconciler) GetLogger(ctx context.Context) logr.Logg
 // +kubebuilder:rbac:groups=ols.openshift.io,resources=olsconfigs/status,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=ols.openshift.io,resources=olsconfigs/finalizers,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=operators.coreos.com,resources=clusterserviceversions,verbs=get;list;
+// +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete;
+// +kubebuilder:rbac:groups="",resources=pods,verbs=create;delete;get;list;patch;update;watch
+// +kubebuilder:rbac:groups="",resources=pods/log,verbs=get
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.
@@ -142,6 +145,19 @@ func (r *OpenStackLightspeedReconciler) Reconcile(ctx context.Context, req ctrl.
 		return ctrl.Result{}, fmt.Errorf("installation of OpenShift LightSpeed not detected: %w", err)
 	}
 
+	indexID, result, err := lightspeed.ResolveIndexID(ctx, helper, instance)
+	if err != nil {
+		instance.Status.Conditions.Set(condition.FalseCondition(
+			lightspeedv1.OpenStackLightspeedReadyCondition,
+			condition.ErrorReason,
+			condition.SeverityWarning,
+			condition.DeploymentReadyErrorMessage,
+			err.Error()))
+		return result, err
+	} else if (result != ctrl.Result{}) {
+		return result, nil
+	}
+
 	// NOTE: We cannot consume the OLSConfig definition directly from the OLS operator's code due to
 	// a conflict in Go versions. When this comment was written, the min. required Go version for
 	// openstack-operator was 1.21 whereas OLS operator required at least Go version 1.23. Once the
@@ -170,7 +186,7 @@ func (r *OpenStackLightspeedReconciler) Reconcile(ctx context.Context, req ctrl.
 			return fmt.Errorf("OLSConfig is managed by different OpenStackLightspeed instance")
 		}
 
-		err = lightspeed.PatchOLSConfig(&olsConfig, instance, helper)
+		err = lightspeed.PatchOLSConfig(helper, instance, &olsConfig, indexID)
 		if err != nil {
 			return err
 		}

pkg/lightspeed/funcs.go

@@ -18,12 +18,21 @@ package lightspeed
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"fmt"
+	"io"
+	batchv1 "k8s.io/api/batch/v1"
+	"k8s.io/client-go/kubernetes"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
 	"strings"
+	"time"
 
 	common_helper "github.com/openstack-k8s-operators/lib-common/modules/common/helper"
 	lightspeedv1beta1 "github.com/openstack-k8s-operators/openstack-operator/apis/lightspeed/v1beta1"
+	corev1 "k8s.io/api/core/v1"
 	k8s_errors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	uns "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -44,6 +53,10 @@ const (
 	// located
 	OpenStackLightspeedVectorDBPath = "/rag/vector_db/os_product_docs"
 
+	// OpenStackLightspeedJobName - name of the pod that is used to discover environment variables inside of the RAG
+	// container image
+	OpenStackLightspeedJobName = "openstack-lightspeed"
+
 	// OLSConfigName - OLS forbids other name for OLSConfig instance than OLSConfigName
 	OLSConfigName = "cluster"
 )
@@ -98,9 +111,10 @@ func IsOLSOperatorInstalled(ctx context.Context, helper *common_helper.Helper) (
 
 // PatchOLSConfig patches OLSConfig with information from OpenStackLightspeed instance.
 func PatchOLSConfig(
-	olsConfig *uns.Unstructured,
-	instance *lightspeedv1beta1.OpenStackLightspeed,
 	helper *common_helper.Helper,
+	instance *lightspeedv1beta1.OpenStackLightspeed,
+	olsConfig *uns.Unstructured,
+	indexID string,
 ) error {
 	// 1. Patch the Providers section
 	providersPatch := []interface{}{
@@ -124,11 +138,6 @@ func PatchOLSConfig(
 		return err
 	}
 
-	indexID, err := getIndexID(instance.Spec.RAGImage)
-	if err != nil {
-		return err
-	}
-
 	// 2. Patch the RAG section
 	openstackRAG := []interface{}{
 		map[string]interface{}{
@@ -143,7 +152,7 @@ func PatchOLSConfig(
 	}
 
 	tlsCaCertBundle := StringPtrValue(instance.Spec.TLSCACertBundle)
-	err = uns.SetNestedField(olsConfig.Object, tlsCaCertBundle, "spec", "ols", "additionalCAConfigMapRef", "name")
+	err := uns.SetNestedField(olsConfig.Object, tlsCaCertBundle, "spec", "ols", "additionalCAConfigMapRef", "name")
 	if err != nil {
 		return err
 	}
@@ -216,15 +225,50 @@ func IsOLSConfigReady(ctx context.Context, helper *common_helper.Helper) (bool,
 	return true, nil
 }
 
-// getIndexID - returns index ID for the data stored in the vector DB container image.
-// It expects that the index ID is equal to the image tag.
-func getIndexID(imageName string) (string, error) {
-	imageNameSections := strings.Split(imageName, ":")
-	if len(imageNameSections) != 2 {
-		return "", fmt.Errorf("failed to discvoer index ID")
+// ResolveIndexID - returns index ID for the data stored in the vector DB container image. The discovery of the
+// index ID is done through spawning a pod with the rag-content image and looking at the INDEX_NAME env variable value.
+func ResolveIndexID(
+	ctx context.Context,
+	helper *common_helper.Helper,
+	instance *lightspeedv1beta1.OpenStackLightspeed,
+) (string, ctrl.Result, error) {
+	result, err := createOLSPod(ctx, helper, instance)
+	if err != nil {
+		return "", result, err
+	}
+
+	podList := &corev1.PodList{}
+	labelSelector := client.MatchingLabels{"app": OpenStackLightspeedJobName}
+	if err := helper.GetClient().List(ctx, podList, client.InNamespace(instance.Namespace), labelSelector); err != nil {
+		return "", ctrl.Result{}, err
 	}
 
-	return imageNameSections[1], nil
+	var OLSPod *corev1.Pod
+	for _, pod := range podList.Items {
+		if pod.Spec.Containers[0].Image == instance.Spec.RAGImage {
+			OLSPod = &pod
+			break
+		}
+	}
+	if OLSPod == nil {
+		helper.GetLogger().Info("waiting for OpenStackLightspeed pod")
+		return "", ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+	}
+
+	switch OLSPod.Status.Phase {
+	case corev1.PodSucceeded:
+		indexName, err := extractEnvFromPodLogs(ctx, OLSPod, "INDEX_NAME")
+		if err != nil && k8s_errors.IsNotFound(err) {
+			helper.GetLogger().Info("waiting for OpenStackLightspeed pod")
+			return "", ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+		}
+		return indexName, ctrl.Result{}, err
+	case corev1.PodFailed:
+		return "", ctrl.Result{}, fmt.Errorf("failed to start OpenStack Lightpseed RAG pod")
+	default:
+		helper.GetLogger().Info("waiting for OpenStackLightspeed pod")
+		return "", ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+	}
 }
 
 // StringPtrValue - dereference safely string pointer
@@ -235,3 +279,102 @@ func StringPtrValue(s *string) string {
 
 	return *s
 }
+
+// extractEnvFromPodLogs - discovers an environment variable value from the pod logs. The pod must be started using
+// createOLSPod.
+func extractEnvFromPodLogs(ctx context.Context, pod *corev1.Pod, envVarName string) (string, error) {
+	cfg, err := config.GetConfig()
+	if err != nil {
+		return "", err
+	}
+
+	k8sClient, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return "", err
+	}
+
+	req := k8sClient.CoreV1().Pods(pod.Namespace).GetLogs(pod.Name, &corev1.PodLogOptions{})
+	podLogs, err := req.Stream(ctx)
+	if err != nil {
+		return "", err
+	}
+	defer podLogs.Close()
+
+	buf := new(strings.Builder)
+	_, err = io.Copy(buf, podLogs)
+	if err != nil {
+		return "", fmt.Errorf("error in copying logs: %w", err)
+	}
+
+	logs := buf.String()
+	for _, envLine := range strings.Split(logs, "\n") {
+		parts := strings.Split(envLine, "=")
+		if len(parts) != 2 {
+			continue
+		}
+
+		if parts[0] == envVarName {
+			return parts[1], nil
+		}
+	}
+
+	return "", fmt.Errorf("env var not discovered: %s", envVarName)
+}
+
+// createOLSPod - starts OLS pod with entrypoint that lists environment variables after the start of the pod. It used
+// to discover INDEX_NAME value.
+func createOLSPod(
+	ctx context.Context,
+	helper *common_helper.Helper,
+	instance *lightspeedv1beta1.OpenStackLightspeed,
+) (ctrl.Result, error) {
+	imageHash := sha256.Sum256([]byte(instance.Spec.RAGImage))
+	imageHashStr := fmt.Sprintf("%x", imageHash)
+	imageHashStr = imageHashStr[len(imageHashStr)-9:]
+	imageName := fmt.Sprintf("%s-%s", OpenStackLightspeedJobName, imageHashStr)
+
+	ttlSecondsAfterFinished := int32(600) // 10 mins
+	activeDeadlineSeconds := int64(1200)  // 20 mins
+	OLSPod := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      imageName,
+			Namespace: instance.Namespace,
+			Labels: map[string]string{
+				"app": OpenStackLightspeedJobName,
+			},
+		},
+		Spec: batchv1.JobSpec{
+			TTLSecondsAfterFinished: &ttlSecondsAfterFinished,
+			ActiveDeadlineSeconds:   &activeDeadlineSeconds,
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: map[string]string{
+						"app": OpenStackLightspeedJobName,
+					},
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:    "rag-content",
+							Image:   instance.Spec.RAGImage,
+							Command: []string{"/bin/sh", "-c"},
+							Args:    []string{"env"},
+						},
+					},
+					RestartPolicy: corev1.RestartPolicyNever,
+				},
+			},
+		},
+	}
+
+	if err := controllerutil.SetControllerReference(instance, OLSPod, helper.GetScheme()); err != nil {
+		return ctrl.Result{}, err
+	}
+
+	err := helper.GetClient().Create(ctx, OLSPod)
+	if err != nil && !k8s_errors.IsAlreadyExists(err) {
+		return ctrl.Result{}, err
+	}
+
+	return ctrl.Result{}, nil
+}