main.go

/*
Copyright 2021.

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 main

import (
	"context"
	"crypto/tls"
	"flag"
	"fmt"
	"os"
	"runtime"
	"time"

	// 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"

	schedmanifests "github.com/openshift-kni/numaresources-operator/pkg/numaresourcesscheduler/manifests/sched"
	securityv1 "github.com/openshift/api/security/v1"
	machineconfigv1 "github.com/openshift/machine-config-operator/pkg/apis/machineconfiguration.openshift.io/v1"
	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
	k8sruntime "k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/klog/v2"
	"k8s.io/klog/v2/klogr"
	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"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
	"sigs.k8s.io/controller-runtime/pkg/webhook"

	"github.com/k8stopologyawareschedwg/deployer/pkg/deployer/platform"
	"github.com/k8stopologyawareschedwg/deployer/pkg/manifests"
	apimanifests "github.com/k8stopologyawareschedwg/deployer/pkg/manifests/api"
	rtemanifests "github.com/k8stopologyawareschedwg/deployer/pkg/manifests/rte"
	depobjupdate "github.com/k8stopologyawareschedwg/deployer/pkg/objectupdate"

	nropv1 "github.com/openshift-kni/numaresources-operator/api/numaresourcesoperator/v1"
	nropv1alpha1 "github.com/openshift-kni/numaresources-operator/api/numaresourcesoperator/v1alpha1"
	"github.com/openshift-kni/numaresources-operator/controllers"
	"github.com/openshift-kni/numaresources-operator/pkg/hash"
	"github.com/openshift-kni/numaresources-operator/pkg/images"
	rteupdate "github.com/openshift-kni/numaresources-operator/pkg/objectupdate/rte"
	schedupdate "github.com/openshift-kni/numaresources-operator/pkg/objectupdate/sched"
	"github.com/openshift-kni/numaresources-operator/pkg/version"
	//+kubebuilder:scaffold:imports
)

const (
	defaultLeaderElectionID = "0e2a6bd3.openshift-kni.io" // autogenerated
)

const (
	defaultWebhookPort = 9443
	defaultMetricsAddr = ":8080"
	defaultProbeAddr   = ":8081"
	defaultImage       = ""
	defaultNamespace   = "numaresources-operator"
)

var (
	scheme = k8sruntime.NewScheme()
)

func init() {
	utilruntime.Must(clientgoscheme.AddToScheme(scheme))

	utilruntime.Must(apiextensionsv1.AddToScheme(scheme))
	utilruntime.Must(nropv1.AddToScheme(scheme))
	utilruntime.Must(nropv1alpha1.AddToScheme(scheme))
	utilruntime.Must(machineconfigv1.Install(scheme))
	utilruntime.Must(securityv1.Install(scheme))
	//+kubebuilder:scaffold:scheme
}

type RenderParams struct {
	NRTCRD         bool
	Namespace      string
	Image          string
	ImageScheduler string
}

type Params struct {
	webhookPort           int
	metricsAddr           string
	enableLeaderElection  bool
	probeAddr             string
	platformName          string
	platformVersion       string
	detectPlatformOnly    bool
	showVersion           bool
	enableScheduler       bool
	renderMode            bool
	render                RenderParams
	enableWebhooks        bool
	enableMetrics         bool
	enableHTTP2           bool
	enableMCPCondsForward bool
}

func (pa *Params) SetDefaults() {
	pa.metricsAddr = defaultMetricsAddr
	pa.probeAddr = defaultProbeAddr
	pa.render.Namespace = defaultNamespace
	pa.render.Image = defaultImage
}

func (pa *Params) FromFlags() {
	flag.StringVar(&pa.metricsAddr, "metrics-bind-address", pa.metricsAddr, "The address the metric endpoint binds to.")
	flag.StringVar(&pa.probeAddr, "health-probe-bind-address", pa.probeAddr, "The address the probe endpoint binds to.")
	flag.BoolVar(&pa.enableLeaderElection, "leader-elect", pa.enableLeaderElection, "Enable leader election for controller manager. Enabling this will ensure there is only one active controller manager.")
	flag.StringVar(&pa.platformName, "platform", pa.platformName, "platform to deploy on - leave empty to autodetect")
	flag.StringVar(&pa.platformVersion, "platform-version", pa.platformVersion, "platform version to deploy on - leave empty to autodetect")
	flag.BoolVar(&pa.detectPlatformOnly, "detect-platform-only", pa.detectPlatformOnly, "detect and report the platform, then exits")
	flag.BoolVar(&pa.renderMode, "render", pa.renderMode, "outputs the rendered manifests, then exits")
	flag.BoolVar(&pa.render.NRTCRD, "render-nrt-crd", pa.render.NRTCRD, "outputs only the rendered NodeResourceTopology CRD manifest, then exits")
	flag.StringVar(&pa.render.Namespace, "render-namespace", pa.render.Namespace, "outputs the manifests rendered using the given namespace")
	flag.StringVar(&pa.render.Image, "render-image", pa.render.Image, "outputs the manifests rendered using the given image")
	flag.StringVar(&pa.render.ImageScheduler, "render-image-scheduler", pa.render.ImageScheduler, "outputs the manifests rendered using the given image for the scheduler")
	flag.BoolVar(&pa.showVersion, "version", pa.showVersion, "outputs the version and exit")
	flag.BoolVar(&pa.enableScheduler, "enable-scheduler", pa.enableScheduler, "enable support for the NUMAResourcesScheduler object")
	flag.BoolVar(&pa.enableWebhooks, "enable-webhooks", pa.enableWebhooks, "enable conversion webhooks")
	flag.IntVar(&pa.webhookPort, "webhook-port", defaultWebhookPort, "The port the operator webhook should listen to.")
	flag.BoolVar(&pa.enableMetrics, "enable-metrics", pa.enableMetrics, "enable metrics server")
	flag.BoolVar(&pa.enableHTTP2, "enable-http2", pa.enableHTTP2, "If HTTP/2 should be enabled for the webhook servers.")
	flag.BoolVar(&pa.enableMCPCondsForward, "enable-mcp-conds-fwd", pa.enableMCPCondsForward, "enable MCP Status Condition forwarding")

	flag.Parse()

	// fix dependant options

	if !pa.enableMetrics {
		pa.metricsAddr = "0"
	}
}

func main() {
	var params Params
	klog.InitFlags(nil)
	params.SetDefaults()
	params.FromFlags()

	if params.showVersion {
		fmt.Printf("%s %s %s %s\n", version.ProgramName(), version.Get(), version.GetGitCommit(), runtime.Version())
		os.Exit(0)
	}

	logh := klogr.NewWithOptions(klogr.WithFormat(klogr.FormatKlog))
	ctrl.SetLogger(logh)

	ctx := context.Background()

	clusterPlatform, clusterPlatformVersion, err := version.DiscoverCluster(ctx, version.ProgramName(), params.platformName, params.platformVersion)
	if err != nil {
		os.Exit(1)
	}

	if params.detectPlatformOnly {
		fmt.Printf("platform=%s version=%s\n", clusterPlatform, clusterPlatformVersion)
		os.Exit(0)
	}

	apiManifests, err := apimanifests.GetManifests(clusterPlatform)
	if err != nil {
		klog.ErrorS(err, "unable to load the API manifests")
		os.Exit(1)
	}
	klog.InfoS("manifests loaded", "component", "API")

	// TODO: we should align image fetch and namespace ENV variables names
	// get the namespace where the operator should install components
	namespace, ok := os.LookupEnv("NAMESPACE")
	if !ok {
		namespace = defaultNamespace
	}

	rteManifests, err := rtemanifests.GetManifests(clusterPlatform, clusterPlatformVersion, namespace, true)
	if err != nil {
		klog.ErrorS(err, "unable to load the RTE manifests")
		os.Exit(1)
	}
	klog.InfoS("manifests loaded", "component", "RTE")

	if params.renderMode {
		os.Exit(manageRendering(params.render, clusterPlatform, apiManifests, rteManifests, namespace, params.enableScheduler))
	}

	klog.InfoS("metrics server", "enabled", params.enableMetrics, "addr", params.metricsAddr)

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Cache:  cache.Options{}, // TODO: restrict namespace here?
		Scheme: scheme,
		Metrics: metricsserver.Options{
			// TODO: secureServing?
			BindAddress: params.metricsAddr,
		},
		WebhookServer: webhook.NewServer(webhook.Options{
			Port:    params.webhookPort,
			TLSOpts: webhookTLSOpts(params.enableHTTP2),
		}),
		HealthProbeBindAddress:  params.probeAddr,
		LeaderElection:          params.enableLeaderElection,
		LeaderElectionNamespace: namespace,
		LeaderElectionID:        defaultLeaderElectionID,
	})
	if err != nil {
		klog.ErrorS(err, "unable to start manager")
		os.Exit(1)
	}

	imageSpec, pullPolicy, err := images.GetCurrentImage(context.Background())
	if err != nil {
		// intentionally continue
		klog.InfoS("unable to find current image, using hardcoded", "error", err)
	}
	klog.InfoS("using RTE image", "spec", imageSpec)

	rteManifestsRendered, err := renderRTEManifests(rteManifests, namespace, imageSpec)
	if err != nil {
		klog.ErrorS(err, "unable to render RTE manifests", "controller", "NUMAResourcesOperator")
		os.Exit(1)
	}

	if err = (&controllers.NUMAResourcesOperatorReconciler{
		Client:          mgr.GetClient(),
		Scheme:          mgr.GetScheme(),
		Recorder:        mgr.GetEventRecorderFor("numaresources-controller"),
		APIManifests:    apiManifests,
		RTEManifests:    rteManifestsRendered,
		Platform:        clusterPlatform,
		ImageSpec:       imageSpec,
		ImagePullPolicy: pullPolicy,
		Namespace:       namespace,
		ForwardMCPConds: params.enableMCPCondsForward,
	}).SetupWithManager(mgr); err != nil {
		klog.ErrorS(err, "unable to create controller", "controller", "NUMAResourcesOperator")
		os.Exit(1)
	}
	if err = (&controllers.KubeletConfigReconciler{
		Client:    mgr.GetClient(),
		Scheme:    mgr.GetScheme(),
		Recorder:  mgr.GetEventRecorderFor("kubeletconfig-controller"),
		Namespace: namespace,
	}).SetupWithManager(mgr); err != nil {
		klog.ErrorS(err, "unable to create controller", "controller", "KubeletConfig")
		os.Exit(1)
	}

	if params.enableScheduler {
		schedMf, err := schedmanifests.GetManifests(namespace)
		if err != nil {
			klog.ErrorS(err, "unable to load the Scheduler manifests")
			os.Exit(1)
		}
		klog.InfoS("manifests loaded", "component", "Scheduler")

		if err = (&controllers.NUMAResourcesSchedulerReconciler{
			Client:             mgr.GetClient(),
			Scheme:             mgr.GetScheme(),
			SchedulerManifests: schedMf,
			Namespace:          namespace,
		}).SetupWithManager(mgr); err != nil {
			klog.ErrorS(err, "unable to create controller", "controller", "NUMAResourcesScheduler")
			os.Exit(1)
		}
	}

	if params.enableWebhooks {
		if err = SetupOperatorWebhookWithManager(mgr, &nropv1.NUMAResourcesOperator{}); err != nil {
			klog.Exitf("unable to create NUMAResourcesOperator v1 webhook : %v", err)
		}
		if err = SetupSchedulerWebhookWithManager(mgr, &nropv1.NUMAResourcesScheduler{}); err != nil {
			klog.Exitf("unable to create NUMAResourcesScheduler v1 webhook : %v", err)
		}
	}

	//+kubebuilder:scaffold:builder

	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		klog.ErrorS(err, "unable to set up health check")
		os.Exit(1)
	}
	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
		klog.ErrorS(err, "unable to set up ready check")
		os.Exit(1)
	}

	klog.InfoS("starting manager")
	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		klog.ErrorS(err, "problem running manager")
		os.Exit(1)
	}
}

func manageRendering(render RenderParams, clusterPlatform platform.Platform, apiMf apimanifests.Manifests, rteMf rtemanifests.Manifests, namespace string, enableScheduler bool) int {
	if render.NRTCRD {
		if err := renderObjects(apiMf.ToObjects()); err != nil {
			klog.ErrorS(err, "unable to render manifests")
			return 1
		}
		return 0
	}

	var objs []client.Object
	if enableScheduler {
		if render.ImageScheduler == "" {
			klog.Errorf("missing scheduler image")
			return 1
		}

		schedMf, err := schedmanifests.GetManifests(namespace)
		if err != nil {
			klog.ErrorS(err, "unable to load the Scheduler manifests")
			return 1
		}
		klog.InfoS("manifests loaded", "component", "Scheduler")

		mf, err := renderSchedulerManifests(schedMf, render.ImageScheduler)
		if err != nil {
			klog.ErrorS(err, "unable to render scheduler manifests")
			return 1
		}
		objs = append(objs, mf.ToObjects()...)
	}

	mf, err := renderRTEManifests(rteMf, render.Namespace, render.Image)
	if err != nil {
		klog.ErrorS(err, "unable to render RTE manifests")
		return 1
	}
	objs = append(objs, mf.ToObjects()...)

	if err := renderObjects(objs); err != nil {
		klog.ErrorS(err, "unable to render manifests")
		return 1
	}

	return 0
}

func renderObjects(objs []client.Object) error {
	for _, obj := range objs {
		fmt.Printf("---\n")
		if err := manifests.SerializeObject(obj, os.Stdout); err != nil {
			return err
		}
	}

	return nil
}

// renderRTEManifests renders the reconciler manifests so they can be deployed on the cluster.
func renderRTEManifests(rteManifests rtemanifests.Manifests, namespace string, imageSpec string) (rtemanifests.Manifests, error) {
	klog.InfoS("Updating RTE manifests")
	mf, err := rteManifests.Render(rtemanifests.RenderOptions{
		Namespace: namespace,
		DaemonSet: depobjupdate.DaemonSetOptions{
			Verbose:            2,
			NotificationEnable: true,
			UpdateInterval:     10 * time.Second,
		},
	})
	if err != nil {
		return mf, err
	}

	err = rteupdate.DaemonSetUserImageSettings(mf.DaemonSet, "", imageSpec, images.NullPolicy)
	if err != nil {
		return mf, err
	}

	err = rteupdate.DaemonSetPauseContainerSettings(mf.DaemonSet)
	if err != nil {
		return mf, err
	}
	if mf.ConfigMap != nil {
		rteupdate.DaemonSetHashAnnotation(mf.DaemonSet, hash.ConfigMapData(mf.ConfigMap))
	}
	return mf, err
}

func renderSchedulerManifests(schedManifests schedmanifests.Manifests, imageSpec string) (schedmanifests.Manifests, error) {
	klog.InfoS("Updating scheduler manifests")
	mf := schedManifests.Clone()
	schedupdate.DeploymentImageSettings(mf.Deployment, imageSpec)
	// empty string is fine. Will be handled as "disabled".
	// We only care about setting the environ variable to declare it exists,
	// the best setting is "present, but disabled" vs "missing, thus implicitly disabled"
	schedupdate.DeploymentConfigMapSettings(mf.Deployment, schedManifests.ConfigMap.Name, hash.ConfigMapData(schedManifests.ConfigMap))
	return mf, nil
}

// SetupWebhookWithManager enables Webhooks - needed for version conversion
func SetupOperatorWebhookWithManager(mgr ctrl.Manager, r *nropv1.NUMAResourcesOperator) error {
	return ctrl.NewWebhookManagedBy(mgr).
		For(r).
		Complete()
}

// SetupWebhookWithManager enables Webhooks - needed for version conversion
func SetupSchedulerWebhookWithManager(mgr ctrl.Manager, r *nropv1.NUMAResourcesScheduler) error {
	return ctrl.NewWebhookManagedBy(mgr).
		For(r).
		Complete()
}

func webhookTLSOpts(enableHTTP2 bool) []func(config *tls.Config) {
	disableHTTP2 := func(c *tls.Config) {
		klog.InfoS("HTTP2 serving for webhook", "enabled", enableHTTP2)
		if enableHTTP2 {
			return
		}
		c.NextProtos = []string{"http/1.1"}
	}

	return []func(config *tls.Config){disableHTTP2}
}