operator.go

package operator

import (
	"context"
	"crypto/tls"
	"fmt"
	"os"
	"path/filepath"
	"slices"
	"time"

	configv1 "github.com/openshift/api/config/v1"
	operatorv1 "github.com/openshift/api/operator/v1"
	openshifttls "github.com/openshift/controller-runtime-common/pkg/tls"
	v1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apiserver/pkg/server/dynamiccertificates"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/cache"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/healthz"
	"sigs.k8s.io/controller-runtime/pkg/manager"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"

	stack "github.com/rhobs/observability-operator/pkg/apis/monitoring/v1alpha1"
	obsv1alpha1 "github.com/rhobs/observability-operator/pkg/apis/observability/v1alpha1"
	uiv1alpha1 "github.com/rhobs/observability-operator/pkg/apis/uiplugin/v1alpha1"
	stackctrl "github.com/rhobs/observability-operator/pkg/controllers/monitoring/monitoring-stack"
	tqctrl "github.com/rhobs/observability-operator/pkg/controllers/monitoring/thanos-querier"
	obsctrl "github.com/rhobs/observability-operator/pkg/controllers/observability"
	opctrl "github.com/rhobs/observability-operator/pkg/controllers/operator"
	uictrl "github.com/rhobs/observability-operator/pkg/controllers/uiplugin"
	ctrlutil "github.com/rhobs/observability-operator/pkg/controllers/util"
)

const (
	// The mount path for the serving certificate seret is hardcoded in the
	// static assets.
	tlsMountPath = "/etc/tls/private"
)

// Operator embeds a manager and a serving certificate controller (for
// OpenShift installations).
type Operator struct {
	manager               manager.Manager
	restConfig            *rest.Config
	servingCertController *dynamiccertificates.DynamicServingCertificateController
	clientCAController    *dynamiccertificates.ConfigMapCAController
}

type OpenShiftFeatureGates struct {
	Enabled bool `json:"enabled,omitempty"`
}

type FeatureGates struct {
	OpenShift OpenShiftFeatureGates `json:"openshift,omitempty"`
}

type OperatorConfiguration struct {
	Namespace              string
	MetricsAddr            string
	HealthProbeAddr        string
	Prometheus             stackctrl.PrometheusConfiguration
	Alertmanager           stackctrl.AlertmanagerConfiguration
	ThanosSidecar          stackctrl.ThanosConfiguration
	ThanosQuerier          tqctrl.ThanosConfiguration
	UIPlugins              uictrl.UIPluginsConfiguration
	FeatureGates           FeatureGates
	ObservabilityInstaller ObservabilityInstallerConfiguration
	TLSProfile             configv1.TLSProfileSpec
	// CancelFunc is called to trigger graceful shutdown (e.g., on TLS profile change).
	CancelFunc context.CancelFunc
}

type ObservabilityInstallerConfiguration struct {
	COONamespace     string
	OpenTelemetryCSV string
	TempoCSV         string
}

func WithNamespace(ns string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.Namespace = ns
		oc.UIPlugins.ResourcesNamespace = ns
	}
}

func WithPrometheusImage(image string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.Prometheus.Image = image
	}
}

func WithAlertmanagerImage(image string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.Alertmanager.Image = image
	}
}

func WithThanosSidecarImage(image string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.ThanosSidecar.Image = image
	}
}

func WithThanosQuerierImage(image string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.ThanosQuerier.Image = image
	}
}

func WithMetricsAddr(addr string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.MetricsAddr = addr
	}
}

func WithHealthProbeAddr(addr string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.HealthProbeAddr = addr
	}
}

func WithUIPluginImages(images map[string]string) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.UIPlugins.Images = images
	}
}

func WithFeatureGates(featureGates FeatureGates) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.FeatureGates = featureGates
	}
}

func NewOperatorConfiguration(opts ...func(*OperatorConfiguration)) *OperatorConfiguration {
	cfg := &OperatorConfiguration{}
	for _, o := range opts {
		o(cfg)
	}
	return cfg
}

func WithObservabilityInstaller(configuration ObservabilityInstallerConfiguration) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.ObservabilityInstaller = configuration
	}
}

func WithCancelFunc(cancel context.CancelFunc) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.CancelFunc = cancel
	}
}

func WithTLSProfile(tlsProfile configv1.TLSProfileSpec) func(*OperatorConfiguration) {
	return func(oc *OperatorConfiguration) {
		oc.TLSProfile = tlsProfile
		oc.UIPlugins.TLSProfile = tlsProfile
	}
}

func New(ctx context.Context, cfg *OperatorConfiguration) (*Operator, error) {
	restConfig := ctrl.GetConfigOrDie()
	scheme := NewScheme(cfg)
	setupLog := ctrl.Log.WithName("setup")

	metricsOpts := metricsserver.Options{
		BindAddress: cfg.MetricsAddr,
	}

	var (
		clientCAController    *dynamiccertificates.ConfigMapCAController
		servingCertController *dynamiccertificates.DynamicServingCertificateController
	)
	if cfg.FeatureGates.OpenShift.Enabled {
		// When running in OpenShift, the server uses HTTPS thanks to the
		// service CA operator.
		certFile := filepath.Join(tlsMountPath, "tls.crt")
		keyFile := filepath.Join(tlsMountPath, "tls.key")

		// Wait for the files to be mounted into the container.
		var pollErr error
		err := wait.PollUntilContextTimeout(ctx, time.Second, 30*time.Second, true, func(ctx context.Context) (bool, error) {
			for _, f := range []string{certFile, keyFile} {
				if _, err := os.Stat(f); err != nil {
					pollErr = err
					return false, nil
				}
			}

			return true, nil
		})
		if err != nil {
			return nil, fmt.Errorf("%w: %w", err, pollErr)
		}

		// DynamicCertKeyPairContent automatically reloads the certificate and key from disk.
		certKeyProvider, err := dynamiccertificates.NewDynamicServingContentFromFiles("serving-cert", certFile, keyFile)
		if err != nil {
			return nil, err
		}
		if err := certKeyProvider.RunOnce(ctx); err != nil {
			return nil, fmt.Errorf("failed to initialize cert/key content: %w", err)
		}

		kubeClient, err := kubernetes.NewForConfig(restConfig)
		if err != nil {
			return nil, err
		}

		clientCAController, err = dynamiccertificates.NewDynamicCAFromConfigMapController(
			"client-ca",
			metav1.NamespaceSystem,
			"extension-apiserver-authentication",
			"client-ca-file",
			kubeClient,
		)
		if err != nil {
			return nil, fmt.Errorf("failed to initialize client CA controller: %w", err)
		}

		// Only log the events emitted by the certificate controller for now
		// because the controller generates invalid events rejected by the
		// Kubernetes API when used with DynamicServingContentFromFiles.
		eventBroadcaster := record.NewBroadcaster()
		eventBroadcaster.StartLogging(func(format string, args ...interface{}) {
			ctrl.Log.WithName("events").Info(fmt.Sprintf(format, args...))
		})

		var tlsConfig tls.Config
		tlsConfigFn, unsupportedCiphers := openshifttls.NewTLSConfigFromProfile(cfg.TLSProfile)
		if len(unsupportedCiphers) > 0 {
			setupLog.Info("Some ciphers from TLS profile are not supported", "ciphers", unsupportedCiphers)
		}
		tlsConfigFn(&tlsConfig)
		tlsConfig.ClientAuth = tls.RequireAndVerifyClientCert

		servingCertController = dynamiccertificates.NewDynamicServingCertificateController(
			&tlsConfig,
			clientCAController,
			certKeyProvider,
			nil,
			record.NewEventRecorderAdapter(
				eventBroadcaster.NewRecorder(scheme, v1.EventSource{Component: "observability-operator"}),
			),
		)
		if err := servingCertController.RunOnce(); err != nil {
			return nil, fmt.Errorf("failed to initialize serving certificate controller: %w", err)
		}

		clientCAController.AddListener(servingCertController)
		certKeyProvider.AddListener(servingCertController)

		metricsOpts.SecureServing = true
		metricsOpts.TLSOpts = []func(*tls.Config){
			func(c *tls.Config) {
				c.GetConfigForClient = servingCertController.GetConfigForClient
			},
		}
	}

	cacheOptions := cache.Options{
		// All controller created resources carry the label
		// defined below. This is added in the reconcilers.
		DefaultLabelSelector: labels.SelectorFromSet(map[string]string{ctrlutil.ResourceLabel: ctrlutil.OpName}),
		ByObject: map[client.Object]cache.ByObject{
			// We define exceptions for the cache and
			// thus from the default label selector.
			// Secrets are watched by some controllers
			// that accept TLS artifacts in a secret.
			&v1.Secret{}: cache.ByObject{
				Label: labels.Everything(),
			},
			// The user-facing CRDs need to be
			// cached in absence of any labels.
			&stack.MonitoringStack{}: cache.ByObject{
				Label: labels.Everything(),
			},
			&stack.ThanosQuerier{}: cache.ByObject{
				Label: labels.Everything(),
			},
			&uiv1alpha1.UIPlugin{}: cache.ByObject{
				Label: labels.Everything(),
			},
			&obsv1alpha1.ObservabilityInstaller{}: cache.ByObject{
				Label: labels.Everything(),
			},
			// The operator controller watches the
			// service created by the olm bundle, so
			// it can create a ServiceMonitor that
			// can be scraped by the OCP in-cluster
			// stack
			&v1.Service{}: cache.ByObject{
				Label: labels.SelectorFromSet(map[string]string{"app.kubernetes.io/name": ctrlutil.OpName}),
			},
		},
	}

	if cfg.FeatureGates.OpenShift.Enabled {
		// APIServer CR is watched for TLS profile changes.
		cacheOptions.ByObject[&configv1.APIServer{}] = cache.ByObject{
			Label: labels.Everything(),
		}
	}

	mgr, err := ctrl.NewManager(
		restConfig,
		ctrl.Options{
			Scheme:                 scheme,
			Metrics:                metricsOpts,
			HealthProbeBindAddress: cfg.HealthProbeAddr,
			PprofBindAddress:       "127.0.0.1:8083",
			Cache:                  cacheOptions,
		})
	if err != nil {
		return nil, fmt.Errorf("unable to create manager: %w", err)
	}

	if err := stackctrl.RegisterWithManager(mgr, stackctrl.Options{
		Prometheus:   cfg.Prometheus,
		Alertmanager: cfg.Alertmanager,
		Thanos:       cfg.ThanosSidecar,
	}); err != nil {
		return nil, fmt.Errorf("unable to register monitoring stack controller: %w", err)
	}

	if err := tqctrl.RegisterWithManager(mgr, tqctrl.Options{Thanos: cfg.ThanosQuerier}); err != nil {
		return nil, fmt.Errorf("unable to register the thanos querier controller with the manager: %w", err)
	}

	if cfg.FeatureGates.OpenShift.Enabled {

		watcher := &openshifttls.SecurityProfileWatcher{
			Client:                mgr.GetClient(),
			InitialTLSProfileSpec: cfg.TLSProfile,
			OnProfileChange: func(_ context.Context, _, _ configv1.TLSProfileSpec) {
				setupLog.Info("TLS security profile changed, triggering graceful restart")
				if cfg.CancelFunc != nil {
					cfg.CancelFunc()
				}
			},
		}
		if err = watcher.SetupWithManager(mgr); err != nil {
			return nil, fmt.Errorf("unable to setup TLS profile watcher: %w", err)
		}

		if err := uictrl.RegisterWithManager(mgr, uictrl.Options{PluginsConf: cfg.UIPlugins}); err != nil {
			return nil, fmt.Errorf("unable to register observability-ui-plugin controller: %w", err)
		}
	} else {
		setupLog.Info("OpenShift feature gate is disabled, UIPlugins are not enabled")
	}

	if cfg.FeatureGates.OpenShift.Enabled {
		if err := opctrl.RegisterWithManager(mgr, cfg.Namespace); err != nil {
			return nil, fmt.Errorf("unable to register operator controller: %w", err)
		}
	} else {
		setupLog.Info("OpenShift feature gate is disabled, Operator controller is not enabled")
	}

	if cfg.FeatureGates.OpenShift.Enabled {
		if err := obsctrl.RegisterWithManager(mgr, obsctrl.Options{
			COONamespace: cfg.ObservabilityInstaller.COONamespace,
			OpenTelemetryOperator: obsctrl.OperatorInstallConfig{
				Namespace:   cfg.ObservabilityInstaller.COONamespace,
				PackageName: "opentelemetry-product",
				StartingCSV: cfg.ObservabilityInstaller.OpenTelemetryCSV,
				Channel:     "stable",
			},
			TempoOperator: obsctrl.OperatorInstallConfig{
				Namespace:   cfg.ObservabilityInstaller.COONamespace,
				PackageName: "tempo-product",
				StartingCSV: cfg.ObservabilityInstaller.TempoCSV,
				Channel:     "stable",
			},
		}); err != nil {
			return nil, fmt.Errorf("unable to register cluster observability controller: %w", err)
		}
	} else {
		setupLog.Info("OpenShift feature gate is disabled, cluster observability controller is not enabled")
	}

	if err := mgr.AddHealthzCheck("health probe", healthz.Ping); err != nil {
		return nil, fmt.Errorf("unable to add health probe: %w", err)
	}

	op := &Operator{
		manager:               mgr,
		restConfig:            restConfig,
		servingCertController: servingCertController,
		clientCAController:    clientCAController,
	}

	if cfg.FeatureGates.OpenShift.Enabled {
		if err := mgr.Add(op.newShutdownCleanupRunnable()); err != nil {
			return nil, fmt.Errorf("unable to add shutdown cleanup runnable: %w", err)
		}
	}

	return op, nil
}

func (o *Operator) Start(ctx context.Context) error {
	if o.clientCAController != nil {
		go o.clientCAController.Run(ctx, 1)
	}

	if o.servingCertController != nil {
		go o.servingCertController.Run(1, ctx.Done())
	}

	if err := o.manager.Start(ctx); err != nil {
		return fmt.Errorf("unable to start manager: %w", err)
	}

	return nil
}

func (o *Operator) newShutdownCleanupRunnable() manager.Runnable {
	return manager.RunnableFunc(func(ctx context.Context) error {
		// Block until the manager's context is cancelled (shutdown signal).
		<-ctx.Done()
		o.cleanupUIPluginsFromConsole()
		return nil
	})
}

func (o *Operator) cleanupUIPluginsFromConsole() {
	logger := ctrl.Log.WithName("shutdown-cleanup")
	logger.Info("attempting best-effort UIPlugin console deregistration")

	cleanupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	directClient, err := client.New(o.restConfig, client.Options{
		Scheme: o.manager.GetScheme(),
	})
	if err != nil {
		logger.Error(err, "failed to create client for shutdown cleanup")
		return
	}

	pluginList := &uiv1alpha1.UIPluginList{}
	if err := directClient.List(cleanupCtx, pluginList); err != nil {
		logger.Error(err, "failed to list UIPlugins during shutdown cleanup")
		return
	}

	if len(pluginList.Items) == 0 {
		return
	}

	toRemove := make(map[string]struct{}, len(pluginList.Items))
	for _, plugin := range pluginList.Items {
		if name := uictrl.ConsoleNameForType(plugin.Spec.Type); name != "" {
			toRemove[name] = struct{}{}
		}
	}

	if len(toRemove) == 0 {
		return
	}

	cluster := &operatorv1.Console{}
	if err := directClient.Get(cleanupCtx, client.ObjectKey{Name: "cluster"}, cluster); err != nil {
		logger.Error(err, "failed to get Console CR during shutdown cleanup")
		return
	}

	original := cluster.DeepCopy()
	cluster.Spec.Plugins = slices.DeleteFunc(cluster.Spec.Plugins, func(name string) bool {
		_, ok := toRemove[name]
		return ok
	})

	if slices.Equal(cluster.Spec.Plugins, original.Spec.Plugins) {
		return
	}

	patch := client.MergeFrom(original)
	if err := directClient.Patch(cleanupCtx, cluster, patch); err != nil {
		logger.Error(err, "failed to patch Console CR during shutdown cleanup")
		return
	}
	logger.Info("successfully cleaned up Console CR during shutdown")
}

func (o *Operator) GetClient() client.Client {
	return o.manager.GetClient()
}