Building ExternalArtifact Controllers

kind create cluster --name dev

flux install \
  --namespace=flux-system \
  --network-policy=false \
  --components=source-controller,kustomize-controller,helm-controller

kubectl -n flux-system patch deployment kustomize-controller \
  --type='json' -p='[{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value": "--feature-gates=ExternalArtifact=true"}]'

kubectl -n flux-system patch deployment helm-controller \
  --type='json' -p='[{"op": "add", "path": "/spec/template/spec/containers/0/args/-", "value": "--feature-gates=ExternalArtifact=true"}]'

git clone https://github.com/fluxcd/source-watcher
cd source-watcher
git checkout v2

go get github.com/fluxcd/pkg/artifact
go get github.com/fluxcd/source-controller/api

import (
    "github.com/spf13/pflag"
    "github.com/fluxcd/pkg/artifact/config"
)

func main() {
    opts := &config.Options{}

    // Bind CLI flags for --storage-path, --storage-addr,
    // --storage-adv-addr, --artifact-retention-ttl,
    // --artifact-retention-records, --artifact-digest-algo.
    opts.BindFlags(pflag.CommandLine)
    pflag.Parse()
}

import (
    "github.com/fluxcd/pkg/artifact/config"
    "github.com/fluxcd/pkg/artifact/storage"
)

// Create storage from configuration options.
store, err := storage.New(opts)
if err != nil {
    panic(err)
}

import (
    "github.com/fluxcd/pkg/artifact/server"
)

// Start the artifact server after the controller-manager receives leadership.
go func() {
    <-mgr.Elected()
    if err := server.Start(ctx, opts); err != nil {
        setupLog.Error(err, "unable to start artifact server")
    }
}()

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "github.com/fluxcd/pkg/apis/meta"
    "github.com/fluxcd/pkg/artifact/storage"
)

func (r *ArtifactGeneratorReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    // Create a new artifact descriptor.
    artifact := store.NewArtifactFor(
        "MySource",                    // kind
        &mySourceObject.ObjectMeta,    // metadata (namespace + name)
        revision,                      // e.g. "v1.0.0@sha256:abc123..."
        fmt.Sprintf("%s.tar.gz", hash),// filename
    )

    // Ensure the artifact directory exists.
    if err := store.MkdirAll(artifact); err != nil {
        return ctrl.Result{}, err
    }

    // Archive a directory into a tarball.
    // The filter excludes .git and other VCS directories.
    if err := store.Archive(&artifact, "/path/to/source/dir", nil); err != nil {
        return ctrl.Result{}, err
    }

    // At this point, artifact.Digest, artifact.Size, and
    // artifact.LastUpdateTime are automatically set by the SDK.
    return ctrl.Result{}, nil
}

import (
    sourcev1 "github.com/fluxcd/source-controller/api/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "sigs.k8s.io/controller-runtime/pkg/client"
)

func (r *ArtifactGeneratorReconciler) reconcileExternalArtifact(ctx context.Context,
    name, namespace string, artifact meta.Artifact) error {

    ea := &sourcev1.ExternalArtifact{
        ObjectMeta: metav1.ObjectMeta{
            Name:      name,
            Namespace: namespace,
        },
    }

    _, err := ctrl.CreateOrUpdate(ctx, r.Client, ea, func() error {
        // Set the artifact status.
        ea.Status.Artifact = &artifact
        // Mark the ExternalArtifact as ready.
        ea.Status.Conditions = []metav1.Condition{
            {
                Type:               "Ready",
                Status:             metav1.ConditionTrue,
                LastTransitionTime: metav1.Now(),
                Reason:             "Succeeded",
                Message:            fmt.Sprintf("stored artifact for revision %s", artifact.Revision),
            },
        }
        return nil
    })
    return err
}

import (
    "time"
    "github.com/fluxcd/pkg/artifact/storage"
)

func (r *ArtifactGeneratorReconciler) garbageCollect(ctx context.Context, artifact meta.Artifact) error {
    // GarbageCollect removes stale artifacts based on the
    // configured retention TTL and max records.
    deleted, err := store.GarbageCollect(ctx, artifact, 5*time.Minute)
    if err != nil {
        return err
    }
    if len(deleted) > 0 {
        log.Info("garbage collected artifacts", "count", len(deleted))
    }
    return nil
}

// Verify that the artifact on disk matches the expected digest.
if err := store.VerifyArtifact(artifact); err != nil {
    log.Error(err, "artifact integrity check failed")
    // Re-fetch or re-generate the artifact.
}

apiVersion: kustomize.toolkit.fluxcd.io/v1
kind: Kustomization
metadata:
  name: my-app
  namespace: apps
spec:
  interval: 10m
  sourceRef:
    kind: ExternalArtifact
    name: my-app
  path: "./"
  prune: true

apiVersion: helm.toolkit.fluxcd.io/v2
kind: HelmRelease
metadata:
  name: my-chart
  namespace: apps
spec:
  interval: 10m
  releaseName: my-chart
  chartRef:
    kind: ExternalArtifact
    name: my-chart

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingAdmissionPolicy
metadata:
  name: "trusted-external-artifacts"
spec:
  failurePolicy: Fail
  matchConstraints:
    resourceRules:
    - apiGroups:   ["source.toolkit.fluxcd.io"]
      apiVersions: ["v1"]
      operations:  ["CREATE", "UPDATE"]
      resources:   ["externalartifacts"]
  validations:
    # Restrict the artifacts to be served only by trusted endpoints
    - expression: >
        !has(object.status.artifact) ||
        object.status.artifact.url.startsWith('http://my-controller.flux-system.svc.cluster.local./')        
    # Restrict the artifact operations to trusted service accounts
    - expression: >
        request.userInfo.username == 'system:serviceaccount:flux-system:my-controller'