[V2] Program controller and file server for program objects (#673)

### Proposed changes

This PR enables a Workspace pod to fetch a Program object in a similar
fashion to how it fetches Flux artifacts. We do this by exposing a HTTP
server that serves tarballs of a fully formed `Pulumi.yaml` file,
incorporating the requested Program spec. Unlike the approach with the
Flux source-controller, I've opted **not to** create/store these tar
files in local ephemeral storage in the controller pod. Instead, when
the artifact URL is accessed, the file server will fetch the requested
Program resource from the Kubernetes API server, and wrap it up in a
`Pulumi.yaml` file and tarred. This approach ensures that the most
recent Program spec is always served, and it also greatly simplifies
storage since we do not need to create a local duplicate of these
Program objects. Since the source of truth is always on cluster, we do
not need to continuously reconcile and generate new artifacts for new
generations of Programs. Should we choose to change this implementation
in the future, the current strategy using the status field to convey the
URL should make it easy to do so.

- [x] Add status field to the Program resource to advertise a
downloadable URL for the program
- [x] Scaffold a program-controller to reconcile the status/URL
- [x] Create a simple file server to serve the fully-formed Pulumi.yaml
from a Program URL
- [x] Update deployment manifests to expose the file server
- [x] Additional unit tests
- [x] Rebase PR to take in test changes
- [x] Integrate with stack-controller

### Related issues (optional)

Closes: #651

operator/Makefile

@@ -144,7 +144,7 @@ build: manifests generate fmt vet ## Build manager binary.
 
 .PHONY: run
 run: manifests generate fmt vet ## Run a controller from your host.
-	go run ./cmd/main.go
+	go run ./cmd/main.go --program-fs-adv-addr=localhost:9090
 
 # If you wish to build the manager image targeting other platforms you can use the --platform flag.
 # (i.e. docker build --platform linux/arm64). However, you must enable docker buildKit for it.

operator/PROJECT

@@ -41,6 +41,7 @@ resources:
 - api:
     crdVersion: v1
     namespaced: true
+  controller: true
   group: pulumi.com
   kind: Program
   path: github.com/pulumi/pulumi-kubernetes-operator/operator/api/pulumi/v1

operator/api/pulumi/v1/artifact_types.go

@@ -0,0 +1,30 @@
+// Copyright 2016-2024, Pulumi Corporation.
+//
+// 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 v1
+
+// Artifact represents the output of a Program reconciliation. This struct is
+// intended to hold information about the programs to be stored.
+type Artifact struct {
+	// URL is the HTTP address of the artifact as exposed by the controller
+	// managing the source spec. It can be used to retrieve the artifact for
+	// consumption, e.g. by another controller applying the artifact contents.
+	// +required
+	URL string `json:"url"`
+
+	// Digest is the digest of the file in the form of '<algorithm>:<checksum>'.
+	// +optional
+	// +kubebuilder:validation:Pattern="^[a-z0-9]+(?:[.+_-][a-z0-9]+)*:[a-zA-Z0-9=_-]+$"
+	Digest string `json:"digest,omitempty"`
+}

operator/api/pulumi/v1/program_types.go

@@ -160,16 +160,31 @@ type Getter struct {
 	State map[string]Expression `json:"state,omitempty"`
 }
 
+// StackStatus defines the observed state of Program.
+type ProgramStatus struct {
+	// ObservedGeneration is the last observed generation of the Program
+	// object.
+	// +optional
+	ObservedGeneration int64 `json:"observedGeneration,omitempty"`
+
+	// Artifact represents the last succesful artifact generated by program reconciliation.
+	// +optional
+	Artifact *Artifact `json:"artifact,omitempty"`
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
+// +kubebuilder:subresource:status
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
+// +kubebuilder:printcolumn:name="URL",type="string",JSONPath=".status.artifact.url"
 
 // Program is the schema for the inline YAML program API.
 type Program struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
 
-	Program ProgramSpec `json:"program,omitempty"`
+	Program ProgramSpec   `json:"program,omitempty"`
+	Status  ProgramStatus `json:"status,omitempty"`
 }
 
 //+kubebuilder:object:root=true

operator/cmd/main.go

@@ -17,20 +17,20 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"crypto/tls"
 	"flag"
+	"net"
+	"net/http"
 	"os"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	sourcev1 "github.com/fluxcd/source-controller/api/v1"
 	sourcev1b2 "github.com/fluxcd/source-controller/api/v1beta2"
-	autov1alpha1 "github.com/pulumi/pulumi-kubernetes-operator/operator/api/auto/v1alpha1"
-	pulumiv1 "github.com/pulumi/pulumi-kubernetes-operator/operator/api/pulumi/v1"
-	autocontroller "github.com/pulumi/pulumi-kubernetes-operator/operator/internal/controller/auto"
-	pulumicontroller "github.com/pulumi/pulumi-kubernetes-operator/operator/internal/controller/pulumi"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -39,6 +39,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
+
+	autov1alpha1 "github.com/pulumi/pulumi-kubernetes-operator/operator/api/auto/v1alpha1"
+	pulumiv1 "github.com/pulumi/pulumi-kubernetes-operator/operator/api/pulumi/v1"
+	autocontroller "github.com/pulumi/pulumi-kubernetes-operator/operator/internal/controller/auto"
+	pulumicontroller "github.com/pulumi/pulumi-kubernetes-operator/operator/internal/controller/pulumi"
 	//+kubebuilder:scaffold:imports
 )
 
@@ -57,11 +62,18 @@ func init() {
 }
 
 func main() {
-	var metricsAddr string
-	var enableLeaderElection bool
-	var probeAddr string
-	var secureMetrics bool
-	var enableHTTP2 bool
+	var (
+		metricsAddr          string
+		enableLeaderElection bool
+		probeAddr            string
+		secureMetrics        bool
+		enableHTTP2          bool
+
+		// Flags for configuring the Program file server.
+		programFSAddr    string
+		programFSAdvAddr string
+	)
+
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
@@ -71,6 +83,10 @@ func main() {
 		"If set the metrics endpoint is served securely")
 	flag.BoolVar(&enableHTTP2, "enable-http2", false,
 		"If set, HTTP/2 will be enabled for the metrics and webhook servers")
+	flag.StringVar(&programFSAddr, "program-fs-addr", envOrDefault("PROGRAM_FS_ADDR", ":9090"),
+		"The address the static file server binds to.")
+	flag.StringVar(&programFSAdvAddr, "program-fs-adv-addr", envOrDefault("PROGRAM_FS_ADV_ADDR", ""),
+		"The advertised address of the static file server.")
 	opts := zap.Options{
 		Development: true,
 	}
@@ -127,6 +143,10 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Create a new ProgramHandler to handle Program objects. Both the ProgramReconciler and the file server need to
+	// access the ProgramHandler, so it is created here and passed to both.
+	pHandler := newProgramHandler(mgr.GetClient(), programFSAdvAddr)
+
 	if err = (&autocontroller.WorkspaceReconciler{
 		Client:   mgr.GetClient(),
 		Scheme:   mgr.GetScheme(),
@@ -151,6 +171,15 @@ func main() {
 		setupLog.Error(err, "unable to create controller", "controller", "Stack")
 		os.Exit(1)
 	}
+	if err = (&pulumicontroller.ProgramReconciler{
+		Client:         mgr.GetClient(),
+		Scheme:         mgr.GetScheme(),
+		Recorder:       mgr.GetEventRecorderFor(pulumicontroller.ProgramControllerName),
+		ProgramHandler: pHandler,
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "Program")
+		os.Exit(1)
+	}
 	//+kubebuilder:scaffold:builder
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
@@ -162,9 +191,92 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Start the file server for serving Program objects.
+	setupLog.Info("starting file server for program resource",
+		"address", programFSAddr,
+		"advertisedAddress", programFSAdvAddr,
+	)
+	if err := mgr.Add(pFileserver{pHandler, programFSAddr}); err != nil {
+		setupLog.Error(err, "unable to start file server for program resource")
+		os.Exit(1)
+	}
+
 	setupLog.Info("starting manager")
 	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
 		setupLog.Error(err, "problem running manager")
 		os.Exit(1)
 	}
 }
+
+// pFileserver implements the manager.Runnable interface to start a simple file server to serve Program objects as
+// compressed tarballs.
+type pFileserver struct {
+	handler *pulumicontroller.ProgramHandler
+	address string
+}
+
+// Start starts the file server to serve Program objects as compressed tarballs.
+func (fs pFileserver) Start(ctx context.Context) error {
+	mux := http.NewServeMux()
+	mux.Handle("/programs/", fs.handler.HandleProgramServing())
+
+	server := &http.Server{
+		Addr:    fs.address,
+		Handler: mux,
+	}
+
+	errChan := make(chan error)
+	go func() {
+		err := server.ListenAndServe()
+		if err != nil {
+			errChan <- err
+		}
+	}()
+
+	select {
+	case err := <-errChan:
+		return err
+	case <-ctx.Done():
+		return server.Shutdown(ctx)
+	}
+}
+
+func newProgramHandler(k8sClient client.Client, advAddr string) *pulumicontroller.ProgramHandler {
+	if advAddr == "" {
+		advAddr = determineAdvAddr(advAddr)
+	}
+
+	return pulumicontroller.NewProgramHandler(k8sClient, advAddr)
+}
+
+func determineAdvAddr(addr string) string {
+	host, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		setupLog.Error(err, "unable to parse file server address")
+		os.Exit(1)
+	}
+	switch host {
+	case "":
+		host = "localhost"
+	case "0.0.0.0":
+		host = os.Getenv("HOSTNAME")
+		if host == "" {
+			hn, err := os.Hostname()
+			if err != nil {
+				setupLog.Error(err, "0.0.0.0 specified in file server addr but hostname is invalid")
+				os.Exit(1)
+			}
+			host = hn
+		}
+	}
+	return net.JoinHostPort(host, port)
+}
+
+func envOrDefault(envName, defaultValue string) string {
+	ret := os.Getenv(envName)
+	if ret != "" {
+		return ret
+	}
+
+	return defaultValue
+}