tatara-operator M1 (Project + Repository reconcilers + ingest Job) Implementation Plan¶

package obs

import (
    "strings"
    "testing"

    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/testutil"
)

func TestReconcileTotal(t *testing.T) {
    reg := prometheus.NewRegistry()
    m := NewOperatorMetrics(reg)

    m.ReconcileResult("Project", "success")
    m.ReconcileResult("Project", "success")
    m.ReconcileResult("Repository", "error")

    got := testutil.ToFloat64(m.reconcileTotal.WithLabelValues("Project", "success"))
    if got != 2 {
        t.Fatalf("Project/success = %v, want 2", got)
    }
    got = testutil.ToFloat64(m.reconcileTotal.WithLabelValues("Repository", "error"))
    if got != 1 {
        t.Fatalf("Repository/error = %v, want 1", got)
    }
}

func TestIngestJobDuration(t *testing.T) {
    reg := prometheus.NewRegistry()
    m := NewOperatorMetrics(reg)

    m.ObserveIngestJobDuration(12.5)

    mfs, err := reg.Gather()
    if err != nil {
        t.Fatalf("gather: %v", err)
    }
    var found bool
    for _, mf := range mfs {
        if mf.GetName() == "operator_ingest_job_duration_seconds" {
            found = true
            if got := mf.GetMetric()[0].GetHistogram().GetSampleCount(); got != 1 {
                t.Fatalf("sample count = %d, want 1", got)
            }
        }
    }
    if !found {
        t.Fatal("operator_ingest_job_duration_seconds not registered")
    }
}

func TestOperatorMetricsNamesStable(t *testing.T) {
    reg := prometheus.NewRegistry()
    _ = NewOperatorMetrics(reg)
    mfs, _ := reg.Gather()
    want := map[string]bool{
        "operator_reconcile_total":             false,
        "operator_ingest_job_duration_seconds": false,
    }
    for _, mf := range mfs {
        if _, ok := want[mf.GetName()]; ok {
            want[mf.GetName()] = true
        }
    }
    for name, seen := range want {
        if !seen {
            t.Errorf("metric %q not registered", name)
        }
    }
    _ = strings.TrimSpace
}

package obs

import "github.com/prometheus/client_golang/prometheus"

// OperatorMetrics holds the reconciler-facing Prometheus collectors for the
// tatara-operator. Construct one with NewOperatorMetrics and pass it to the
// reconcilers.
type OperatorMetrics struct {
    reconcileTotal    *prometheus.CounterVec
    ingestJobDuration prometheus.Histogram
}

// NewOperatorMetrics registers the operator collectors on reg and returns the
// bundle. Names and labels are pinned by the shared-contracts pin set.
func NewOperatorMetrics(reg prometheus.Registerer) *OperatorMetrics {
    m := &OperatorMetrics{
        reconcileTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
            Name: "operator_reconcile_total",
            Help: "Total reconcile outcomes by kind and result.",
        }, []string{"kind", "result"}),
        ingestJobDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
            Name:    "operator_ingest_job_duration_seconds",
            Help:    "Wall-clock duration of completed ingest Jobs.",
            Buckets: prometheus.ExponentialBuckets(5, 2, 8),
        }),
    }
    reg.MustRegister(m.reconcileTotal, m.ingestJobDuration)
    return m
}

// ReconcileResult increments operator_reconcile_total for the given kind and
// result ("success" or "error").
func (m *OperatorMetrics) ReconcileResult(kind, result string) {
    m.reconcileTotal.WithLabelValues(kind, result).Inc()
}

// ObserveIngestJobDuration records the wall-clock seconds a completed ingest
// Job took.
func (m *OperatorMetrics) ObserveIngestJobDuration(seconds float64) {
    m.ingestJobDuration.Observe(seconds)
}

package ingest

import (
    "strings"
    "testing"

    tataradevv1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func testProject() *tataradevv1alpha1.Project {
    p := &tataradevv1alpha1.Project{}
    p.Name = "acme"
    p.Namespace = "tatara"
    p.Spec.ScmSecretRef = "acme-scm"
    return p
}

func testRepository() *tataradevv1alpha1.Repository {
    r := &tataradevv1alpha1.Repository{}
    r.Name = "widgets"
    r.Namespace = "tatara"
    r.UID = "repo-uid-123"
    r.Spec.ProjectRef = "acme"
    r.Spec.URL = "https://github.com/acme/widgets.git"
    r.Spec.DefaultBranch = "main"
    return r
}

func testConfig() Config {
    return Config{
        IngesterImage:    "registry.example/ingester:1.2.3",
        MemoryBaseURL:    "http://tatara-memory.tatara.svc:8080",
        OIDCIssuer:       "https://kc.example/realms/tatara",
        OIDCClientID:     "tatara-operator",
        OIDCClientSecret: "s3cr3t",
        OIDCAudience:     "tatara-memory",
        Namespace:        "tatara",
    }
}

func envValue(c corev1.Container, key string) string {
    for _, e := range c.Env {
        if e.Name == key {
            return e.Value
        }
    }
    return ""
}

func TestBuildJob_FullIngest(t *testing.T) {
    job := BuildJob(testProject(), testRepository(), "", testConfig())

    if job.Namespace != "tatara" {
        t.Errorf("namespace = %q, want tatara", job.Namespace)
    }
    if !strings.HasPrefix(job.Name, "widgets-ingest-") {
        t.Errorf("job name = %q, want prefix widgets-ingest-", job.Name)
    }
    if got := job.Spec.Template.Spec.RestartPolicy; got != corev1.RestartPolicyNever {
        t.Errorf("restartPolicy = %q, want Never", got)
    }
    if job.Spec.BackoffLimit == nil || *job.Spec.BackoffLimit != 2 {
        t.Errorf("backoffLimit = %v, want 2", job.Spec.BackoffLimit)
    }

    if len(job.OwnerReferences) != 1 {
        t.Fatalf("ownerReferences = %d, want 1", len(job.OwnerReferences))
    }
    or := job.OwnerReferences[0]
    if or.Kind != "Repository" || or.Name != "widgets" || string(or.UID) != "repo-uid-123" {
        t.Errorf("ownerRef = %+v, want Repository/widgets/repo-uid-123", or)
    }
    if or.Controller == nil || !*or.Controller {
        t.Error("ownerRef.Controller should be true")
    }

    initCs := job.Spec.Template.Spec.InitContainers
    if len(initCs) != 1 {
        t.Fatalf("init containers = %d, want 1", len(initCs))
    }
    clone := initCs[0]
    cloneCmd := strings.Join(clone.Command, " ") + " " + strings.Join(clone.Args, " ")
    if !strings.Contains(cloneCmd, "https://github.com/acme/widgets.git") {
        t.Errorf("clone cmd missing url: %q", cloneCmd)
    }
    if !strings.Contains(cloneCmd, "--branch main") {
        t.Errorf("clone cmd missing branch: %q", cloneCmd)
    }

    cs := job.Spec.Template.Spec.Containers
    if len(cs) != 1 {
        t.Fatalf("containers = %d, want 1", len(cs))
    }
    main := cs[0]
    if main.Image != "registry.example/ingester:1.2.3" {
        t.Errorf("image = %q, want registry.example/ingester:1.2.3", main.Image)
    }

    cmd := strings.Join(main.Command, " ") + " " + strings.Join(main.Args, " ")
    if !strings.Contains(cmd, "tatara-ingest --repo-root /workspace/repo --repo-name widgets --base-url http://tatara-memory.tatara.svc:8080") {
        t.Errorf("ingest cmd wrong: %q", cmd)
    }
    if strings.Contains(cmd, "--since") {
        t.Errorf("full ingest must not pass --since: %q", cmd)
    }
    if !strings.Contains(cmd, "widgets-ingest-result") {
        t.Errorf("ingest cmd must write result configmap: %q", cmd)
    }

    if v := envValue(main, "BASE_URL"); v != "http://tatara-memory.tatara.svc:8080" {
        t.Errorf("BASE_URL = %q", v)
    }
    if v := envValue(main, "OIDC_ISSUER"); v != "https://kc.example/realms/tatara" {
        t.Errorf("OIDC_ISSUER = %q", v)
    }
    if v := envValue(main, "OIDC_CLIENT_ID"); v != "tatara-operator" {
        t.Errorf("OIDC_CLIENT_ID = %q", v)
    }
    if v := envValue(main, "OIDC_CLIENT_SECRET"); v != "s3cr3t" {
        t.Errorf("OIDC_CLIENT_SECRET = %q", v)
    }
    if v := envValue(main, "OIDC_AUDIENCE"); v != "tatara-memory" {
        t.Errorf("OIDC_AUDIENCE = %q", v)
    }
}

func TestBuildJob_IncrementalIngest(t *testing.T) {
    job := BuildJob(testProject(), testRepository(), "abc1234", testConfig())
    main := job.Spec.Template.Spec.Containers[0]
    cmd := strings.Join(main.Command, " ") + " " + strings.Join(main.Args, " ")
    if !strings.Contains(cmd, "--since abc1234") {
        t.Errorf("incremental ingest must pass --since abc1234: %q", cmd)
    }
}

func TestBuildJob_SCMTokenFromSecret(t *testing.T) {
    job := BuildJob(testProject(), testRepository(), "", testConfig())
    clone := job.Spec.Template.Spec.InitContainers[0]
    var ref *corev1.EnvVarSource
    for _, e := range clone.Env {
        if e.Name == "SCM_TOKEN" {
            ref = e.ValueFrom
        }
    }
    if ref == nil || ref.SecretKeyRef == nil {
        t.Fatal("clone container must source SCM_TOKEN from a secret")
    }
    if ref.SecretKeyRef.Name != "acme-scm" || ref.SecretKeyRef.Key != "token" {
        t.Errorf("SCM_TOKEN secretKeyRef = %s/%s, want acme-scm/token",
            ref.SecretKeyRef.Name, ref.SecretKeyRef.Key)
    }
}

func TestBuildJob_SharedWorkspaceVolume(t *testing.T) {
    job := BuildJob(testProject(), testRepository(), "", testConfig())
    ps := job.Spec.Template.Spec
    var hasEmptyDir bool
    for _, v := range ps.Volumes {
        if v.Name == "workspace" && v.EmptyDir != nil {
            hasEmptyDir = true
        }
    }
    if !hasEmptyDir {
        t.Error("pod must have an emptyDir volume named workspace")
    }
    mounted := func(c corev1.Container) bool {
        for _, m := range c.VolumeMounts {
            if m.Name == "workspace" && m.MountPath == "/workspace" {
                return true
            }
        }
        return false
    }
    if !mounted(ps.InitContainers[0]) {
        t.Error("init container must mount workspace at /workspace")
    }
    if !mounted(ps.Containers[0]) {
        t.Error("main container must mount workspace at /workspace")
    }
    _ = metav1.Now
    _ = batchv1.Job{}
}

// Package ingest builds the Kubernetes Job that clones a repository and runs
// tatara-ingest against tatara-memory.
package ingest

import (
    "fmt"

    tataradevv1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/util/rand"
)

// Config is the subset of operator configuration the Job builder needs.
type Config struct {
    IngesterImage    string
    MemoryBaseURL    string
    OIDCIssuer       string
    OIDCClientID     string
    OIDCClientSecret string
    OIDCAudience     string
    Namespace        string
}

// ResultConfigMapName returns the name of the ConfigMap an ingest Job patches
// with the resolved HEAD SHA for the given Repository.
func ResultConfigMapName(repo *tataradevv1alpha1.Repository) string {
    return repo.Name + "-ingest-result"
}

const (
    workspaceVolume = "workspace"
    workspaceMount  = "/workspace"
    repoDir         = "/workspace/repo"
)

// BuildJob returns the *batchv1.Job that ingests repo for project. When since
// is non-empty the ingest is incremental (--since since); otherwise it is a
// full ingest. The Job is owner-referenced to repo. It clones with the
// Project SCM token in an init container into an emptyDir, runs tatara-ingest
// in the main container, then writes the cloned HEAD SHA into the repo's
// result ConfigMap via the in-cluster API.
func BuildJob(project *tataradevv1alpha1.Project, repo *tataradevv1alpha1.Repository, since string, cfg Config) *batchv1.Job {
    backoff := int32(2)
    ttl := int32(3600)
    controller := true

    cloneCmd := fmt.Sprintf(
        "set -e; git clone --depth 1 --branch %s "+
            "https://x-access-token:${SCM_TOKEN}@${REPO_HOST_PATH} %s",
        repo.Spec.DefaultBranch, repoDir)

    ingestArgs := fmt.Sprintf(
        "tatara-ingest --repo-root %s --repo-name %s --base-url %s",
        repoDir, repo.Name, cfg.MemoryBaseURL)
    if since != "" {
        ingestArgs += " --since " + since
    }
    // After a successful ingest, resolve HEAD and patch the result ConfigMap
    // via the in-cluster API (the Job ServiceAccount has patch on it).
    resultCM := ResultConfigMapName(repo)
    mainScript := fmt.Sprintf(
        "set -e; %s; "+
            "SHA=$(git -C %s rev-parse HEAD); "+
            "kubectl -n %s patch configmap %s --type merge "+
            "-p \"{\\\"data\\\":{\\\"sha\\\":\\\"${SHA}\\\"}}\"",
        ingestArgs, repoDir, cfg.Namespace, resultCM)

    repoHostPath := repo.Spec.URL
    repoHostPath = trimScheme(repoHostPath)

    return &batchv1.Job{
        ObjectMeta: metav1.ObjectMeta{
            Name:      repo.Name + "-ingest-" + rand.String(5),
            Namespace: cfg.Namespace,
            Labels: map[string]string{
                "app.kubernetes.io/name":      "tatara-operator",
                "app.kubernetes.io/component": "ingest",
                "tatara.dev/managed-by":       "tatara-operator",
                "tatara.dev/repository":       repo.Name,
            },
            OwnerReferences: []metav1.OwnerReference{{
                APIVersion: tataradevv1alpha1.GroupVersion.String(),
                Kind:       "Repository",
                Name:       repo.Name,
                UID:        repo.UID,
                Controller: &controller,
            }},
        },
        Spec: batchv1.JobSpec{
            BackoffLimit:            &backoff,
            TTLSecondsAfterFinished: &ttl,
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{
                        "app.kubernetes.io/name":      "tatara-operator",
                        "app.kubernetes.io/component": "ingest",
                        "tatara.dev/managed-by":       "tatara-operator",
                        "tatara.dev/repository":       repo.Name,
                    },
                },
                Spec: corev1.PodSpec{
                    RestartPolicy:      corev1.RestartPolicyNever,
                    ServiceAccountName: "tatara-ingest",
                    Volumes: []corev1.Volume{{
                        Name:         workspaceVolume,
                        VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}},
                    }},
                    InitContainers: []corev1.Container{{
                        Name:    "clone",
                        Image:   cfg.IngesterImage,
                        Command: []string{"/bin/sh", "-c"},
                        Args:    []string{cloneCmd},
                        Env: []corev1.EnvVar{
                            {
                                Name: "SCM_TOKEN",
                                ValueFrom: &corev1.EnvVarSource{
                                    SecretKeyRef: &corev1.SecretKeySelector{
                                        LocalObjectReference: corev1.LocalObjectReference{Name: project.Spec.ScmSecretRef},
                                        Key:                  "token",
                                    },
                                },
                            },
                            {Name: "REPO_HOST_PATH", Value: repoHostPath},
                        },
                        VolumeMounts: []corev1.VolumeMount{{Name: workspaceVolume, MountPath: workspaceMount}},
                    }},
                    Containers: []corev1.Container{{
                        Name:    "ingest",
                        Image:   cfg.IngesterImage,
                        Command: []string{"/bin/sh", "-c"},
                        Args:    []string{mainScript},
                        Env: []corev1.EnvVar{
                            {Name: "BASE_URL", Value: cfg.MemoryBaseURL},
                            {Name: "OIDC_ISSUER", Value: cfg.OIDCIssuer},
                            {Name: "OIDC_CLIENT_ID", Value: cfg.OIDCClientID},
                            {Name: "OIDC_CLIENT_SECRET", Value: cfg.OIDCClientSecret},
                            {Name: "OIDC_AUDIENCE", Value: cfg.OIDCAudience},
                        },
                        VolumeMounts: []corev1.VolumeMount{{Name: workspaceVolume, MountPath: workspaceMount}},
                    }},
                },
            },
        },
    }
}

// trimScheme removes a leading https:// or http:// from a URL so it can be
// reassembled with embedded credentials.
func trimScheme(u string) string {
    for _, p := range []string{"https://", "http://"} {
        if len(u) >= len(p) && u[:len(p)] == p {
            return u[len(p):]
        }
    }
    return u
}

package controller

import (
    "context"
    "path/filepath"
    "testing"
    "time"

    tataradevv1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    corev1 "k8s.io/api/core/v1"
    "k8s.io/client-go/kubernetes/scheme"
    "k8s.io/client-go/rest"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/envtest"
)

var (
    testEnv   *envtest.Environment
    cfg       *rest.Config
    k8sClient client.Client
)

const (
    timeout  = 10 * time.Second
    interval = 250 * time.Millisecond
    testNS   = "tatara"
)

// TestMain boots a single envtest control plane for the whole controller
// package, registers the tatara.dev scheme and core types, creates the test
// namespace, and tears the control plane down at the end.
func TestMain(m *testing.M) {
    code := func() int {
        testEnv = &envtest.Environment{
            CRDDirectoryPaths:     []string{filepath.Join("..", "..", "config", "crd", "bases")},
            ErrorIfCRDPathMissing: true,
        }
        var err error
        cfg, err = testEnv.Start()
        if err != nil {
            panic("start envtest: " + err.Error())
        }
        defer func() { _ = testEnv.Stop() }()

        if err := tataradevv1alpha1.AddToScheme(scheme.Scheme); err != nil {
            panic("add scheme: " + err.Error())
        }

        k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
        if err != nil {
            panic("new client: " + err.Error())
        }

        ns := &corev1.Namespace{}
        ns.Name = testNS
        if err := k8sClient.Create(context.Background(), ns); err != nil {
            panic("create namespace: " + err.Error())
        }

        return m.Run()
    }()
    osExit(code)
}

package controller

import "os"

// osExit is a thin wrapper so TestMain has a single exit point.
var osExit = os.Exit

package controller

import (
    "context"
    "testing"
    "time"

    tataradevv1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "github.com/szymonrychu/tatara-operator/internal/obs"
    "github.com/prometheus/client_golang/prometheus"
    corev1 "k8s.io/api/core/v1"
    apierrors "k8s.io/apimachinery/pkg/api/meta"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/types"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    logf "sigs.k8s.io/controller-runtime/pkg/log"
)

func newProjectReconciler() *ProjectReconciler {
    return &ProjectReconciler{
        Client:              k8sClient,
        Scheme:              k8sClient.Scheme(),
        Metrics:             obs.NewOperatorMetrics(prometheus.NewRegistry()),
        ExternalWebhookBase: "https://tatara.example/operator/webhooks",
    }
}

func reconcileProject(t *testing.T, name string) (ctrl.Result, error) {
    t.Helper()
    r := newProjectReconciler()
    return r.Reconcile(logf.IntoContext(context.Background(), logf.Log), ctrl.Request{
        NamespacedName: types.NamespacedName{Namespace: testNS, Name: name},
    })
}

func mkSecret(t *testing.T, name string, data map[string][]byte) {
    t.Helper()
    s := &corev1.Secret{}
    s.Name = name
    s.Namespace = testNS
    s.Data = data
    if err := k8sClient.Create(context.Background(), s); err != nil {
        t.Fatalf("create secret %s: %v", name, err)
    }
}

func getProject(t *testing.T, name string) *tataradevv1alpha1.Project {
    t.Helper()
    p := &tataradevv1alpha1.Project{}
    if err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: testNS, Name: name}, p); err != nil {
        t.Fatalf("get project %s: %v", name, err)
    }
    return p
}

func waitProjectReady(t *testing.T, name string, want metav1.ConditionStatus) *tataradevv1alpha1.Project {
    t.Helper()
    deadline := time.Now().Add(timeout)
    for time.Now().Before(deadline) {
        p := getProject(t, name)
        c := apierrors.FindStatusCondition(p.Status.Conditions, "Ready")
        if c != nil && c.Status == want {
            return p
        }
        time.Sleep(interval)
    }
    t.Fatalf("project %s Ready never reached %s", name, want)
    return nil
}

func TestProjectReconcile_ValidSecret(t *testing.T) {
    ctx := context.Background()
    mkSecret(t, "valid-scm", map[string][]byte{
        "token":         []byte("ghp_x"),
        "webhookSecret": []byte("hmac"),
    })
    p := &tataradevv1alpha1.Project{}
    p.Name = "proj-valid"
    p.Namespace = testNS
    p.Spec.ScmSecretRef = "valid-scm"
    if err := k8sClient.Create(ctx, p); err != nil {
        t.Fatalf("create project: %v", err)
    }

    if _, err := reconcileProject(t, "proj-valid"); err != nil {
        t.Fatalf("reconcile: %v", err)
    }

    got := waitProjectReady(t, "proj-valid", metav1.ConditionTrue)
    want := "https://tatara.example/operator/webhooks/proj-valid"
    if got.Status.WebhookURL != want {
        t.Errorf("webhookURL = %q, want %q", got.Status.WebhookURL, want)
    }
}

func TestProjectReconcile_MissingSecret(t *testing.T) {
    ctx := context.Background()
    p := &tataradevv1alpha1.Project{}
    p.Name = "proj-nosecret"
    p.Namespace = testNS
    p.Spec.ScmSecretRef = "does-not-exist"
    if err := k8sClient.Create(ctx, p); err != nil {
        t.Fatalf("create project: %v", err)
    }

    if _, err := reconcileProject(t, "proj-nosecret"); err != nil {
        t.Fatalf("reconcile returned error, want nil (status carries failure): %v", err)
    }
    got := waitProjectReady(t, "proj-nosecret", metav1.ConditionFalse)
    c := apierrors.FindStatusCondition(got.Status.Conditions, "Ready")
    if c.Reason != "SecretNotFound" {
        t.Errorf("reason = %q, want SecretNotFound", c.Reason)
    }
}

func TestProjectReconcile_MissingKeys(t *testing.T) {
    ctx := context.Background()
    mkSecret(t, "partial-scm", map[string][]byte{"token": []byte("ghp_x")})
    p := &tataradevv1alpha1.Project{}
    p.Name = "proj-partialkeys"
    p.Namespace = testNS
    p.Spec.ScmSecretRef = "partial-scm"
    if err := k8sClient.Create(ctx, p); err != nil {
        t.Fatalf("create project: %v", err)
    }

    if _, err := reconcileProject(t, "proj-partialkeys"); err != nil {
        t.Fatalf("reconcile: %v", err)
    }
    got := waitProjectReady(t, "proj-partialkeys", metav1.ConditionFalse)
    c := apierrors.FindStatusCondition(got.Status.Conditions, "Ready")
    if c.Reason != "SecretMissingKeys" {
        t.Errorf("reason = %q, want SecretMissingKeys", c.Reason)
    }
}

var _ = client.IgnoreNotFound

// Package controller holds the tatara-operator reconcilers.
package controller

import (
    "context"
    "fmt"

    tataradevv1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "github.com/szymonrychu/tatara-operator/internal/obs"
    corev1 "k8s.io/api/core/v1"
    apierrors "k8s.io/apimachinery/pkg/api/errors"
    "k8s.io/apimachinery/pkg/api/meta"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/types"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"
)

// ProjectReconciler validates a Project's SCM secret and publishes its
// webhook URL.
type ProjectReconciler struct {
    client.Client
    Scheme              *runtime.Scheme
    Metrics             *obs.OperatorMetrics
    ExternalWebhookBase string
}

// +kubebuilder:rbac:groups=tatara.dev,resources=projects,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=tatara.dev,resources=projects/status,verbs=get;update;patch
// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch

// Reconcile validates spec.scmSecretRef and sets status.webhookURL plus the
// Ready condition. A missing or malformed secret is reported via the Ready
// condition (status False), not returned as an error.
func (r *ProjectReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    l := log.FromContext(ctx)

    var project tataradevv1alpha1.Project
    if err := r.Get(ctx, req.NamespacedName, &project); err != nil {
        if apierrors.IsNotFound(err) {
            return ctrl.Result{}, nil
        }
        r.Metrics.ReconcileResult("Project", "error")
        return ctrl.Result{}, fmt.Errorf("get project: %w", err)
    }

    reason, message, ready := r.validateSecret(ctx, &project)

    project.Status.WebhookURL = fmt.Sprintf("%s/%s", r.ExternalWebhookBase, project.Name)
    status := metav1.ConditionTrue
    if !ready {
        status = metav1.ConditionFalse
    }
    meta.SetStatusCondition(&project.Status.Conditions, metav1.Condition{
        Type:               "Ready",
        Status:             status,
        Reason:             reason,
        Message:            message,
        ObservedGeneration: project.Generation,
    })

    if err := r.Status().Update(ctx, &project); err != nil {
        r.Metrics.ReconcileResult("Project", "error")
        return ctrl.Result{}, fmt.Errorf("update project status: %w", err)
    }

    l.Info("reconciled project",
        "action", "reconcile_project",
        "resource_id", project.Name,
        "ready", ready,
        "reason", reason)
    r.Metrics.ReconcileResult("Project", "success")
    return ctrl.Result{}, nil
}

// validateSecret returns the condition (reason, message, ready) for the
// Project's scmSecretRef. ready is true only when the secret exists and has
// both required keys.
func (r *ProjectReconciler) validateSecret(ctx context.Context, project *tataradevv1alpha1.Project) (reason, message string, ready bool) {
    if project.Spec.ScmSecretRef == "" {
        return "SecretRefEmpty", "spec.scmSecretRef is empty", false
    }
    var secret corev1.Secret
    key := types.NamespacedName{Namespace: project.Namespace, Name: project.Spec.ScmSecretRef}
    if err := r.Get(ctx, key, &secret); err != nil {
        if apierrors.IsNotFound(err) {
            return "SecretNotFound", fmt.Sprintf("secret %q not found", project.Spec.ScmSecretRef), false
        }
        return "SecretError", err.Error(), false
    }
    for _, k := range []string{"token", "webhookSecret"} {
        if len(secret.Data[k]) == 0 {
            return "SecretMissingKeys", fmt.Sprintf("secret %q missing key %q", project.Spec.ScmSecretRef, k), false
        }
    }
    return "Validated", "scm secret present with token and webhookSecret", true
}

// SetupWithManager registers the reconciler with the manager, watching
// Projects.
func (r *ProjectReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&tataradevv1alpha1.Project{}).
        Owns(&corev1.Secret{}).
        Complete(r)
}

package controller

import (
    "context"
    "testing"
    "time"

    tataradevv1alpha1 "github.com/szymonrychu/tatara-operator/api/v1alpha1"
    "github.com/szymonrychu/tatara-operator/internal/ingest"
    "github.com/szymonrychu/tatara-operator/internal/obs"
    "github.com/prometheus/client_golang/prometheus"
    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/types"
    ctrl "sigs.k8s.io/controller-runtime"
    logf "sigs.k8s.io/controller-runtime/pkg/log"
)

func newRepoReconciler() *RepositoryReconciler {
    return &RepositoryReconciler{
        Client:  k8sClient,
        Scheme:  k8sClient.Scheme(),
        Metrics: obs.NewOperatorMetrics(prometheus.NewRegistry()),
        IngestConfig: ingest.Config{
            IngesterImage: "registry.example/ingester:1.2.3",
            MemoryBaseURL: "http://tatara-memory.tatara.svc:8080",
            OIDCIssuer:    "https://kc.example/realms/tatara",
            OIDCClientID:  "tatara-operator",
            OIDCAudience:  "tatara-memory",
            Namespace:     testNS,
        },
    }
}

func reconcileRepo(t *testing.T, name string) (ctrl.Result, error) {
    t.Helper()
    r := newRepoReconciler()
    return r.Reconcile(logf.IntoContext(context.Background(), logf.Log), ctrl.Request{
        NamespacedName: types.NamespacedName{Namespace: testNS, Name: name},
    })
}

func mkProject(t *testing.T, name, secretRef string) {
    t.Helper()
    p := &tataradevv1alpha1.Project{}
    p.Name = name
    p.Namespace = testNS
    p.Spec.ScmSecretRef = secretRef
    if err := k8sClient.Create(context.Background(), p); err != nil {
        t.Fatalf("create project %s: %v", name, err)
    }
}

func mkRepo(t *testing.T, name, projectRef string) *tataradevv1alpha1.Repository {
    t.Helper()
    r := &tataradevv1alpha1.Repository{}
    r.Name = name
    r.Namespace = testNS
    r.Spec.ProjectRef = projectRef
    r.Spec.URL = "https://github.com/acme/" + name + ".git"
    r.Spec.DefaultBranch = "main"
    r.Spec.IngestEnabled = true
    if err := k8sClient.Create(context.Background(), r); err != nil {
        t.Fatalf("create repo %s: %v", name, err)
    }
    return r
}

func getRepo(t *testing.T, name string) *tataradevv1alpha1.Repository {
    t.Helper()
    r := &tataradevv1alpha1.Repository{}
    if err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: testNS, Name: name}, r); err != nil {
        t.Fatalf("get repo %s: %v", name, err)
    }
    return r
}

func listIngestJobs(t *testing.T, repoName string) []batchv1.Job {
    t.Helper()
    var jl batchv1.JobList
    if err := k8sClient.List(context.Background(), &jl,
        client.InNamespace(testNS),
        client.MatchingLabels{"tatara.dev/repository": repoName}); err != nil {
        t.Fatalf("list jobs: %v", err)
    }
    return jl.Items
}

func waitRepoJob(t *testing.T, repoName string) string {
    t.Helper()
    deadline := time.Now().Add(timeout)
    for time.Now().Before(deadline) {
        r := getRepo(t, repoName)
        if r.Status.JobName != "" {
            return r.Status.JobName
        }
        time.Sleep(interval)
    }
    t.Fatalf("repo %s never set status.jobName", repoName)
    return ""
}

func TestRepoReconcile_FullIngestLaunchesJob(t *testing.T) {
    mkProject(t, "rp-full", "rp-full-scm")
    mkSecret(t, "rp-full-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    mkRepo(t, "full", "rp-full")

    if _, err := reconcileRepo(t, "full"); err != nil {
        t.Fatalf("reconcile: %v", err)
    }
    jobName := waitRepoJob(t, "full")

    jobs := listIngestJobs(t, "full")
    if len(jobs) != 1 {
        t.Fatalf("jobs = %d, want 1", len(jobs))
    }
    if jobs[0].Name != jobName {
        t.Errorf("status.jobName = %q, job = %q", jobName, jobs[0].Name)
    }
    // full ingest: no --since in the main container script
    script := jobs[0].Spec.Template.Spec.Containers[0].Args[0]
    if contains(script, "--since") {
        t.Errorf("full ingest job must not pass --since: %q", script)
    }
    // result ConfigMap pre-created
    cm := &corev1.ConfigMap{}
    if err := k8sClient.Get(context.Background(),
        types.NamespacedName{Namespace: testNS, Name: "full-ingest-result"}, cm); err != nil {
        t.Fatalf("result configmap not pre-created: %v", err)
    }
    if getRepo(t, "full").Status.Phase != "Ingesting" {
        t.Errorf("phase = %q, want Ingesting", getRepo(t, "full").Status.Phase)
    }
}

func TestRepoReconcile_ConcurrencyGuard(t *testing.T) {
    mkProject(t, "rp-guard", "rp-guard-scm")
    mkSecret(t, "rp-guard-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    mkRepo(t, "guard", "rp-guard")

    if _, err := reconcileRepo(t, "guard"); err != nil {
        t.Fatalf("first reconcile: %v", err)
    }
    first := waitRepoJob(t, "guard")

    // second reconcile while the Job is still active must not launch another
    if _, err := reconcileRepo(t, "guard"); err != nil {
        t.Fatalf("second reconcile: %v", err)
    }
    jobs := listIngestJobs(t, "guard")
    if len(jobs) != 1 {
        t.Fatalf("jobs after second reconcile = %d, want 1 (guard held)", len(jobs))
    }
    if getRepo(t, "guard").Status.JobName != first {
        t.Errorf("jobName changed under guard: %q -> %q", first, getRepo(t, "guard").Status.JobName)
    }
}

func TestRepoReconcile_IncrementalUsesSince(t *testing.T) {
    mkProject(t, "rp-inc", "rp-inc-scm")
    mkSecret(t, "rp-inc-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    r := mkRepo(t, "inc", "rp-inc")

    // simulate a prior successful ingest
    r = getRepo(t, "inc")
    r.Status.LastIngestedCommit = "oldsha99"
    r.Status.LastIngestTime = metav1.NewTime(time.Now().Add(-1 * time.Hour))
    r.Status.Phase = "Ingested"
    if err := k8sClient.Status().Update(context.Background(), r); err != nil {
        t.Fatalf("seed status: %v", err)
    }
    // request a re-ingest via the annotation, newer than lastIngestTime
    r = getRepo(t, "inc")
    if r.Annotations == nil {
        r.Annotations = map[string]string{}
    }
    r.Annotations["tatara.dev/reingest-requested"] = time.Now().Format(time.RFC3339)
    if err := k8sClient.Update(context.Background(), r); err != nil {
        t.Fatalf("set annotation: %v", err)
    }

    if _, err := reconcileRepo(t, "inc"); err != nil {
        t.Fatalf("reconcile: %v", err)
    }
    waitRepoJob(t, "inc")
    jobs := listIngestJobs(t, "inc")
    if len(jobs) != 1 {
        t.Fatalf("jobs = %d, want 1", len(jobs))
    }
    script := jobs[0].Spec.Template.Spec.Containers[0].Args[0]
    if !contains(script, "--since oldsha99") {
        t.Errorf("incremental job must pass --since oldsha99: %q", script)
    }
}

func TestRepoReconcile_NoReingestWhenAnnotationStale(t *testing.T) {
    mkProject(t, "rp-stale", "rp-stale-scm")
    mkSecret(t, "rp-stale-scm", map[string][]byte{"token": []byte("x"), "webhookSecret": []byte("y")})
    r := mkRepo(t, "stale", "rp-stale")

    r = getRepo(t, "stale")
    r.Status.LastIngestedCommit = "shaA"
    r.Status.LastIngestTime = metav1.NewTime(time.Now())
    r.Status.Phase = "Ingested"
    if err := k8sClient.Status().Update(context.Background(), r); err != nil {
        t.Fatalf("seed status: %v", err)
    }
    r = getRepo(t, "stale")
    if r.Annotations == nil {
        r.Annotations = map[string]string{}
    }
    // annotation OLDER than lastIngestTime -> no new ingest
    r.Annotations["tatara.dev/reingest-requested"] = time.Now().Add(-2 * time.Hour).Format(time.RFC3339)
    if err := k8sClient.Update(context.Background(), r); err != nil {
        t.Fatalf("set annotation: %v", err)
    }

    if _, err := reconcileRepo(t, "stale"); err != nil {
        t.Fatalf("reconcile: %v", err)
    }
    jobs := listIngestJobs(t, "stale")
    if len(jobs) != 0 {
        t.Fatalf("stale annotation must not launch a job, got %d", len(jobs))
    }
}

func contains(s, sub string) bool {
    return len(sub) == 0 || (len(s) >= len(sub) && indexOf(s, sub) >= 0)
}

func indexOf(s, sub string) int {
    for i := 0; i+len(sub) <= len(s); i++ {
        if s[i:i+len(sub)] == sub {
            return i
        }
    }
    return -1
}