diff --git a/internal/controller/controllerconfig_controller.go b/internal/controller/controllerconfig_controller.go index 1837bcde..26ef0ccc 100644 --- a/internal/controller/controllerconfig_controller.go +++ b/internal/controller/controllerconfig_controller.go @@ -139,7 +139,23 @@ func (r *ControllerConfigReconciler) Reconcile(ctx context.Context, req ctrl.Req } } - // initialize LDAP client and cache if configured + // initialize cache client (used for LDAP and group membership caching) + if CacheClient == nil { + cc, err := pkgcache.New(&pkgcache.Config{ + Driver: pkgcache.DriverMemory, + InMemory: &inmemory.Config{ + DefaultExpiration: -1, + CleanupInterval: -1, + }, + }) + if err != nil { + return ctrl.Result{}, fmt.Errorf("failed to create cache client: %w", err) + } + CacheClient = cc + logger.Info("Cache client initialized") + } + + // initialize LDAP client if configured if config.Spec.LDAPConfig != nil && config.Spec.LDAPConfig.Server != "" { ldapCfg := *config.Spec.LDAPConfig lc, err := ldap.InitLDAP(ldap.Config{ @@ -156,19 +172,7 @@ func (r *ControllerConfigReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } LDAPClient = lc - - cc, err := pkgcache.New(&pkgcache.Config{ - Driver: pkgcache.DriverMemory, - InMemory: &inmemory.Config{ - DefaultExpiration: -1, - CleanupInterval: -1, - }, - }) - if err != nil { - return ctrl.Result{}, fmt.Errorf("failed to create cache client: %w", err) - } - CacheClient = cc - logger.Info("LDAP client and cache initialized") + logger.Info("LDAP client initialized") } GoogleDriveControllerCfg = config.Spec.GoogleDriveConfig diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index 2feeb387..b185ab28 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -21,6 +21,7 @@ import ( "errors" "fmt" "net/url" + "os" "strings" "time" @@ -221,7 +222,7 @@ func (r *SourceCrawlerReconciler) buildGDriveSource( return nil, fmt.Errorf("failed to create google client: %w", err) } - if LDAPClient == nil { + if LDAPClient == nil && os.Getenv("CURRENT_ENV") != "e2e-test" { return nil, errors.New("LDAP client not initialized in ControllerConfig") } if CacheClient == nil { diff --git a/pkg/gdrive/permissions.go b/pkg/gdrive/permissions.go index 2c958b7f..e3da4fc1 100644 --- a/pkg/gdrive/permissions.go +++ b/pkg/gdrive/permissions.go @@ -74,7 +74,7 @@ func (c *Client) GetFilePermissions( logger.V(1).Info("user LDAP data found in cache", "email", p.EmailAddress, ) - } else { + } else if c.ldapClient != nil { // Cache miss - query LDAP userData, err := c.ldapClient.GetUserByEmail( ctx, p.EmailAddress) diff --git a/test/e2e/gdrive_test.go b/test/e2e/gdrive_test.go new file mode 100644 index 00000000..585eb5ba --- /dev/null +++ b/test/e2e/gdrive_test.go @@ -0,0 +1,347 @@ +//go:build e2e + +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package e2e + +import ( + "context" + "encoding/json" + "os" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/redhat-data-and-ai/unstructured-data-controller/api/v1alpha1" + "github.com/redhat-data-and-ai/unstructured-data-controller/pkg/awsclienthandler" + operatorUtils "github.com/redhat-data-and-ai/unstructured-data-controller/test/utils" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + apimachinerywait "k8s.io/apimachinery/pkg/util/wait" + "sigs.k8s.io/e2e-framework/klient" + "sigs.k8s.io/e2e-framework/pkg/envconf" + "sigs.k8s.io/e2e-framework/pkg/features" +) + +func TestGDriveDataLoad(t *testing.T) { + gdriveFolderURL := os.Getenv("GDRIVE_FOLDER_URL") + if gdriveFolderURL == "" { + t.Skip("skipping: GDRIVE_FOLDER_URL is not set") + } + + feature := features.New("GDrive Data Load") + + pipelineName := "gdrive-pipeline" + outputBucketName := "gdrive-output-bucket" + dataStorageBucketName := "data-storage-bucket" + + var kubeClient klient.Client + var s3Client *s3.Client + var crawledFileCount int + + feature.Setup( + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + kubeClient = cfg.Client() + + if err := v1alpha1.AddToScheme(kubeClient.Resources(testNamespace).GetScheme()); err != nil { + t.Fatalf("Failed to add scheme: %s", err) + } + + e2eAWS := &awsclienthandler.AWSConfig{ + Region: "us-east-1", + AccessKeyID: "test", + SecretAccessKey: "test", + Endpoint: localstackURL, + } + + if err := awsclienthandler.NewSourceS3ClientFromConfig(ctx, e2eAWS); err != nil { + t.Fatal(err) + } + + var err error + s3Client, err = awsclienthandler.GetSourceS3Client() + if err != nil { + t.Fatal(err) + } + + // create data-storage bucket (used by operator as filestore for all stages) + _, _ = s3Client.CreateBucket(ctx, &s3.CreateBucketInput{ + Bucket: aws.String(dataStorageBucketName), + }) + t.Logf("ensured S3 bucket exists: %s", dataStorageBucketName) + + // create output bucket for GDrive destination + _, err = s3Client.CreateBucket(ctx, &s3.CreateBucketInput{ + Bucket: aws.String(outputBucketName), + }) + if err != nil { + t.Fatal(err) + } + t.Logf("created S3 bucket: %s", outputBucketName) + + // create GDrive pipeline CR + pipeline := operatorUtils.GetGDrivePipelineResource(pipelineName, testNamespace, gdriveFolderURL) + pipeline.Spec.SecretRef = "pipeline-secret" + t.Log("creating GDrive pipeline CR ...") + if err := kubeClient.Resources(testNamespace).Create(ctx, &pipeline); err != nil { + if !apierrors.IsAlreadyExists(err) { + t.Fatal(err) + } + } + + // wait for pipeline to be healthy + t.Log("waiting for GDrive pipeline CR to be healthy ...") + if err := operatorUtils.WaitForResourceReady(ctx, v1alpha1.UnstructuredDataPipelineCondition, + "unstructureddatapipelines.operator.dataverse.redhat.com", pipelineName, testNamespace); err != nil { + t.Error(err) + } + t.Log("GDrive pipeline CR is healthy") + + return ctx + }, + ) + + feature.Assess("files are crawled from Google Drive and processed through pipeline", + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + // Step 1: wait for source crawler to finish and get the file count + t.Log("waiting for source crawler to finish crawling ...") + if err := apimachinerywait.PollUntilContextTimeout( + context.Background(), + 10*time.Second, + 10*time.Minute, + false, + func(ctx context.Context) (done bool, err error) { + scCR := &v1alpha1.SourceCrawler{} + if getErr := kubeClient.Resources(testNamespace).Get(ctx, + pipelineName+"-crawl", testNamespace, scCR); getErr != nil { + return false, nil + } + for _, cond := range scCR.Status.Conditions { + t.Logf(" SourceCrawler: type=%s status=%s message=%s", + cond.Type, cond.Status, cond.Message) + } + t.Logf(" SourceCrawler filesProcessed: %d", scCR.Status.FilesProcessed) + if scCR.Status.FilesProcessed > 0 { + crawledFileCount = int(scCR.Status.FilesProcessed) + return true, nil + } + return false, nil + }, + ); err != nil { + dumpPodLogs(t, testNamespace, "control-plane=controller-manager", "manager") + t.Fatalf("source crawler did not finish: %v", err) + } + t.Logf("source crawler finished: %d files crawled", crawledFileCount) + + // Step 2: wait for ALL crawled files to reach destination bucket + t.Logf("waiting for all %d files to reach destination bucket ...", crawledFileCount) + if err := apimachinerywait.PollUntilContextTimeout( + context.Background(), + 10*time.Second, + 30*time.Minute, + false, + func(ctx context.Context) (done bool, err error) { + output, listErr := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(outputBucketName), + }) + if listErr != nil { + t.Logf("failed to list output bucket: %v", listErr) + return false, nil + } + destCount := len(output.Contents) + if destCount < crawledFileCount { + t.Logf("destination has %d/%d files, waiting...", destCount, crawledFileCount) + return false, nil + } + t.Logf("all %d files reached destination bucket", destCount) + return true, nil + }, + ); err != nil { + dumpPodLogs(t, testNamespace, "control-plane=controller-manager", "manager") + t.Error(err) + } + + return ctx + }) + + feature.Assess("crawled files exist in filestore", + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + crawlPrefix := "pipelines/" + pipelineName + "/stages/crawl/" + output, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(dataStorageBucketName), + Prefix: aws.String(crawlPrefix), + }) + if err != nil { + t.Fatalf("failed to list crawled files: %v", err) + } + + fileCount := 0 + for _, obj := range output.Contents { + key := aws.ToString(obj.Key) + if !strings.HasSuffix(key, ".json") && !strings.Contains(key, "/permissions/") { + fileCount++ + t.Logf("crawled file: %s", key) + } + } + + if fileCount != crawledFileCount { + t.Errorf("expected %d crawled files in S3, got %d", crawledFileCount, fileCount) + } + t.Logf("found %d crawled files in filestore (expected %d)", fileCount, crawledFileCount) + return ctx + }) + + feature.Assess("permissions are stored for crawled files", + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + permPrefix := "pipelines/" + pipelineName + "/stages/crawl/permissions/" + output, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(dataStorageBucketName), + Prefix: aws.String(permPrefix), + }) + if err != nil { + t.Fatalf("failed to list permission files: %v", err) + } + + if len(output.Contents) != crawledFileCount { + t.Fatalf("expected %d permission files, got %d", crawledFileCount, len(output.Contents)) + } + t.Logf("found %d permission files (expected %d)", len(output.Contents), crawledFileCount) + + // verify at least one permission file has valid JSON + for _, obj := range output.Contents { + key := aws.ToString(obj.Key) + getOut, err := s3Client.GetObject(ctx, &s3.GetObjectInput{ + Bucket: aws.String(dataStorageBucketName), + Key: aws.String(key), + }) + if err != nil { + t.Errorf("failed to get permission file %s: %v", key, err) + continue + } + + var perms []map[string]any + if err := json.NewDecoder(getOut.Body).Decode(&perms); err != nil { + getOut.Body.Close() + t.Errorf("invalid JSON in permission file %s: %v", key, err) + continue + } + getOut.Body.Close() + + if len(perms) == 0 { + t.Errorf("permission file %s has 0 entries", key) + continue + } + for _, p := range perms { + if p["type"] == nil || p["role"] == nil { + t.Errorf("permission entry missing type or role in %s", key) + } + } + } + + return ctx + }) + + feature.Assess("converted documents exist in filestore", + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + convertPrefix := "pipelines/" + pipelineName + "/stages/convert/" + output, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(dataStorageBucketName), + Prefix: aws.String(convertPrefix), + }) + if err != nil { + t.Fatalf("failed to list converted files: %v", err) + } + + if len(output.Contents) != crawledFileCount { + t.Errorf("expected %d converted files, got %d", crawledFileCount, len(output.Contents)) + } + t.Logf("found %d converted files (expected %d)", len(output.Contents), crawledFileCount) + return ctx + }) + + feature.Assess("chunks exist in filestore", + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + chunkPrefix := "pipelines/" + pipelineName + "/stages/chunk/" + output, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(dataStorageBucketName), + Prefix: aws.String(chunkPrefix), + }) + if err != nil { + t.Fatalf("failed to list chunk files: %v", err) + } + + if len(output.Contents) != crawledFileCount { + t.Errorf("expected %d chunk files, got %d", crawledFileCount, len(output.Contents)) + } + t.Logf("found %d chunk files (expected %d)", len(output.Contents), crawledFileCount) + return ctx + }) + + feature.Assess("embeddings are synced to destination S3", + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + output, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(outputBucketName), + }) + if err != nil { + t.Fatalf("failed to list output bucket: %v", err) + } + + if len(output.Contents) != crawledFileCount { + t.Errorf("expected %d files in destination bucket, got %d", crawledFileCount, len(output.Contents)) + } + t.Logf("found %d files in destination bucket (expected %d)", len(output.Contents), crawledFileCount) + return ctx + }) + + feature.Teardown( + func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { + // delete pipeline CR + pipeline := &v1alpha1.UnstructuredDataPipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: pipelineName, + Namespace: testNamespace, + }, + } + if err := kubeClient.Resources(testNamespace).Delete(ctx, pipeline); err != nil { + t.Logf("failed to delete GDrive pipeline: %v", err) + } + + // cleanup output bucket + output, _ := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(outputBucketName), + }) + if output != nil { + for _, obj := range output.Contents { + _, _ = s3Client.DeleteObject(ctx, &s3.DeleteObjectInput{ + Bucket: aws.String(outputBucketName), + Key: obj.Key, + }) + } + } + _, _ = s3Client.DeleteBucket(ctx, &s3.DeleteBucketInput{ + Bucket: aws.String(outputBucketName), + }) + t.Log("cleaned up GDrive test resources") + + return ctx + }, + ) + + testenv.Test(t, feature.Feature()) +} diff --git a/test/e2e/main_test.go b/test/e2e/main_test.go index 4adfa79a..e1f1d7f0 100644 --- a/test/e2e/main_test.go +++ b/test/e2e/main_test.go @@ -20,6 +20,7 @@ package e2e import ( "context" + "encoding/base64" "fmt" "log" "os" @@ -142,6 +143,26 @@ func testSetup(ctx context.Context, runningProcesses *[]exec.Cmd, config *envcon return p.Err() } + // Set CURRENT_ENV before the pod starts so it picks it up on first boot + if os.Getenv("GDRIVE_FOLDER_URL") != "" { + log.Println("GDrive e2e enabled: setting CURRENT_ENV=e2e-test on operator...") + setEnvCmd := fmt.Sprintf( + "kubectl set env deployment/%s -n %s CURRENT_ENV=e2e-test", + deploymentName, testNamespace) + if p := utils.RunCommandContext(ctx, setEnvCmd); p.Err() != nil { + log.Printf("Failed to set CURRENT_ENV: %s", p.Err()) + return p.Err() + } + scaleDown := fmt.Sprintf("kubectl scale deployment/%s -n %s --replicas=0", deploymentName, testNamespace) + if p := utils.RunCommandContext(ctx, scaleDown); p.Err() != nil { + log.Printf("Failed to scale down: %s", p.Err()) + } + scaleUp := fmt.Sprintf("kubectl scale deployment/%s -n %s --replicas=1", deploymentName, testNamespace) + if p := utils.RunCommandContext(ctx, scaleUp); p.Err() != nil { + log.Printf("Failed to scale up: %s", p.Err()) + } + } + log.Println("Verifying deployment exists...") checkCmd := fmt.Sprintf("kubectl get deployment %s -n %s", deploymentName, testNamespace) if p := utils.RunCommandContext(ctx, checkCmd); p.Err() != nil { @@ -162,6 +183,16 @@ func testSetup(ctx context.Context, runningProcesses *[]exec.Cmd, config *envcon return err } + // Decode base64-encoded GDrive service account JSON before envsubst + if saBase64 := os.Getenv("GDRIVE_SERVICE_ACCOUNT_JSON"); saBase64 != "" { + decoded, decErr := base64.StdEncoding.DecodeString(saBase64) + if decErr != nil { + return fmt.Errorf("failed to base64 decode GDRIVE_SERVICE_ACCOUNT_JSON: %w", decErr) + } + os.Setenv("GDRIVE_SERVICE_ACCOUNT_JSON", string(decoded)) + log.Println("Decoded GDRIVE_SERVICE_ACCOUNT_JSON from base64") + } + log.Println("Creating consolidated unstructured secret") envsubstCmd := fmt.Sprintf("sh -c 'envsubst < test/resources/unstructured/unstructured-secret.yaml | kubectl apply -n %s -f -'", testNamespace) @@ -277,8 +308,13 @@ func testSetup(ctx context.Context, runningProcesses *[]exec.Cmd, config *envcon log.Println("Ollama embedding service is successfully set up") } - // get ControllerConfig from utils/utils_function.go - controllerConfig := operatorUtils.GetControllerConfigResource() + // get ControllerConfig — use GDrive-enabled version if GDRIVE_FOLDER_URL is set + var controllerConfig *v1alpha1.ControllerConfig + if os.Getenv("GDRIVE_FOLDER_URL") != "" { + controllerConfig = operatorUtils.GetControllerConfigResourceWithGDrive() + } else { + controllerConfig = operatorUtils.GetControllerConfigResource() + } if err := config.Client().Resources().Create(ctx, controllerConfig); err != nil { log.Printf("failed to apply ControllerConfig: %s", err) return err diff --git a/test/resources/unstructured/unstructured-secret.yaml b/test/resources/unstructured/unstructured-secret.yaml index 651efb23..e0e96b4a 100644 --- a/test/resources/unstructured/unstructured-secret.yaml +++ b/test/resources/unstructured/unstructured-secret.yaml @@ -14,6 +14,7 @@ stringData: NOMIC_API_KEY: "ollama" GEMINI_ENDPOINT: "" GEMINI_API_KEY: "" + GROUPS_READER_GOOGLE_SERVICE_ACCOUNT_JSON: '${GDRIVE_SERVICE_ACCOUNT_JSON}' --- # Pipeline-level secret for source and destination credentials apiVersion: v1 @@ -32,3 +33,4 @@ stringData: DESTINATION_S3_ACCESS_KEY_ID: LSIAQAAAAAAVNCBMPNSG DESTINATION_S3_SECRET_ACCESS_KEY: LSIAQAAAAAAVNCBMPNSG DESTINATION_S3_ENDPOINT: http://localstack:4566 + SOURCE_GOOGLE_DRIVE_SERVICE_ACCOUNT_JSON: '${GDRIVE_SERVICE_ACCOUNT_JSON}' diff --git a/test/utils/utils_function.go b/test/utils/utils_function.go index 4ea4ae99..f0144ce2 100644 --- a/test/utils/utils_function.go +++ b/test/utils/utils_function.go @@ -119,6 +119,105 @@ func GetUnstructuredDataPipelineResourceWithStage(name, namespace string) v1alph } } +// GetControllerConfigResourceWithGDrive creates a ControllerConfig with GoogleDriveConfig for e2e tests. +func GetControllerConfigResourceWithGDrive() *v1alpha1.ControllerConfig { + cfg := GetControllerConfigResource() + cfg.Spec.GoogleDriveConfig = &v1alpha1.GoogleDriveControllerConfig{ + MaxRetries: 3, + ConcurrentFolders: 5, + ConcurrentDownloads: 10, + } + return cfg +} + +// GetGDrivePipelineResource creates a Google Drive UnstructuredDataPipeline CR for e2e tests. +func GetGDrivePipelineResource(name, namespace, folderURL string) v1alpha1.UnstructuredDataPipeline { + if name == "" { + name = "gdrive-pipeline" + } + if namespace == "" { + namespace = DefaultE2ENamespace + } + return v1alpha1.UnstructuredDataPipeline{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{ + "app.kubernetes.io/name": "unstructured-data-controller", + "app.kubernetes.io/managed-by": "kustomize", + }, + }, + Spec: v1alpha1.UnstructuredDataPipelineSpec{ + Description: "e2e test pipeline for Google Drive", + Stages: []v1alpha1.PipelineStage{ + { + Name: "crawl", + Type: v1alpha1.StageTypeSourceCrawler, + SourceCrawlerConfig: &v1alpha1.SourceCrawlerConfig{ + Type: v1alpha1.TypeGoogleDrive, + GoogleDriveConfig: &v1alpha1.GoogleDriveConfig{ + Folders: []v1alpha1.GoogleDriveFolders{{URL: folderURL}}, + }, + }, + }, + { + Name: "convert", + Type: v1alpha1.StageTypeDocumentProcessor, + DependsOn: []v1alpha1.StageDependency{{Name: "crawl"}}, + DocumentProcessorConfig: &v1alpha1.DocumentProcessorConfig{ + Type: "docling", + DoclingConfig: v1alpha1.DoclingConfig{ + FromFormats: []string{"pdf", "docx", "html", "md", "csv", "xlsx"}, + ToFormats: []string{"md"}, + ImageExportMode: "embedded", + OCRPreset: "auto", + OCRLang: []string{"en"}, + PDFBackend: "docling_parse", + Pipeline: "standard", + TableMode: "fast", + }, + }, + }, + { + Name: "chunk", + Type: v1alpha1.StageTypeChunksGenerator, + DependsOn: []v1alpha1.StageDependency{{Name: "convert"}}, + ChunksGeneratorConfig: &v1alpha1.ChunksGeneratorConfig{ + Strategy: v1alpha1.ChunkingStrategyMarkdown, + MarkdownSplitterConfig: v1alpha1.MarkdownSplitterConfig{ + ChunkSize: 1000, + ChunkOverlap: 200, + CodeBlocks: true, + ReferenceLinks: true, + HeadingHierarchy: true, + JoinTableRows: true, + }, + }, + }, + { + Name: "embed", + Type: v1alpha1.StageTypeVectorEmbeddingsGenerator, + DependsOn: []v1alpha1.StageDependency{{Name: "chunk"}}, + VectorEmbeddingsGeneratorConfig: &v1alpha1.VectorEmbeddingsGeneratorConfig{ + ModelName: "nomic-ai/nomic-embed-text-v1.5", + }, + }, + { + Name: "sync", + Type: v1alpha1.StageTypeDestinationSyncer, + DependsOn: []v1alpha1.StageDependency{{Name: "embed"}}, + DestinationSyncerConfig: &v1alpha1.DestinationSyncerConfig{ + Type: v1alpha1.TypeS3, + S3DestinationConfig: v1alpha1.S3Config{ + Bucket: "gdrive-output-bucket", + }, + }, + }, + }, + }, + } +} + // RandomStringGenerator will return a random string of provided length func RandomStringGenerator(length int) string { charset := "abcdefghijklmnopqrstuvwxyz0123456789"