main.go

/*
Copyright 2022 Ondat.

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"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"os"
	"path"
	"path/filepath"
	"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.

	"go.uber.org/zap/zapcore"
	_ "k8s.io/client-go/plugin/pkg/client/auth"
	"k8s.io/client-go/tools/leaderelection/resourcelock"

	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	"k8s.io/apimachinery/pkg/runtime"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/yaml"
	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/util/workqueue"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/healthz"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"

	stosv1 "github.com/storageos/api-manager/api/v1"
	stosoperatorv1 "github.com/storageos/operator/apis/v1"
	storageosv1alpha1 "github.com/storageos/portal-manager/api/v1alpha1"
	"github.com/storageos/portal-manager/controllers"
	"github.com/storageos/portal-manager/endpoints"
	"github.com/storageos/portal-manager/managers"
	"github.com/storageos/portal-manager/pkg/action"
	"github.com/storageos/portal-manager/pkg/action/licence"
	"github.com/storageos/portal-manager/pkg/action/state"
	"github.com/storageos/portal-manager/pkg/handler"
	"github.com/storageos/portal-manager/pkg/publisher"
	"github.com/storageos/portal-manager/pkg/utils"
	klog "k8s.io/klog/v2"
	//+kubebuilder:scaffold:imports
)

const (
	// podNamespace is the operator's pod namespace environment variable.
	podNamespace = "POD_NAMESPACE"

	// portalURLEnv is the Portal API tenant ID environment variable.
	tenantIDEnv = "TENANT_ID"

	// portalURLEnv is the Portal API URL environment variable.
	portalURLEnv = "URL"

	// clientIDEnv is the Portal API client ID environment variable.
	clientIDEnv = "CLIENT_ID"

	// passwordEnv is the Portal API passwordEnv environment variable.
	passwordEnv = "PASSWORD"
)

const (
	// stosServiceTemplate is the name of StorageOS API service.
	stosServiceTemplate = "storageos.%s.svc"

	// stosSecretPath is the path of the mounted API secrets.
	//nolint:gosec // It doesn't contain secret
	stosSecretPath = "/etc/storageos/secrets/api"

	// stosUsernameKey is the key in the secret that holds the username value.
	stosUsernameKey = "username"

	// stosPasswordKey is the key in the secret that holds the password value.
	stosPasswordKey = "password"
)

var (
	scheme              = runtime.NewScheme()
	shutDownperiod      = time.Second * 5
	leaderRetryDuration = 5 * time.Second
)

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

	utilruntime.Must(storageosv1alpha1.AddToScheme(scheme))

	utilruntime.Must(stosv1.AddToScheme(scheme))

	utilruntime.Must(stosoperatorv1.AddToScheme(scheme))
	//+kubebuilder:scaffold:scheme
}

// +kubebuilder:rbac:groups="",resources=namespaces,verbs=get;watch;list
// +kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=configmaps/status,verbs=get;update;patch
// +kubebuilder:rbac:groups="",resources=secrets,verbs=get;create;update;patch;list;watch
// +kubebuilder:rbac:groups="storageos.com",resources=storageosportals,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="storageos.com",resources=storageosclusters,verbs=watch;list
// +kubebuilder:rbac:groups="coordination.k8s.io",resources=leases,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups="",resources=nodes;persistentvolumeclaims;persistentvolumes;pods,verbs=watch;list
// +kubebuilder:rbac:groups="storage.k8s.io",resources=storageclasses;volumeattachments,verbs=watch;list
// +kubebuilder:rbac:groups=apps,resources=statefulsets;daemonsets;deployments;replicasets,verbs=watch;list
// +kubebuilder:rbac:groups="api.storageos.com",resources=volumes;nodes,verbs=watch;list
// +kubebuilder:rbac:groups="",resources=events,verbs=create
// +kubebuilder:rbac:groups=extensions,resources=podsecuritypolicies,verbs=use

// TODO some retries would be nice at every level.

//nolint:gocyclo // It is complex
func main() {
	ctx, cancel := context.WithCancel(ctrl.SetupSignalHandler())
	defer func() {
		cancel()

		time.Sleep(shutDownperiod)

		os.Exit(1)
	}()

	var configFile string
	flag.StringVar(&configFile, "config", "",
		"The controller will load its initial configuration from this file. "+
			"Omit this flag to use the default configuration values. "+
			"Command-line flags override configuration from this file.")

	var leaderRenewSeconds uint
	flag.UintVar(&leaderRenewSeconds, "leader-renew-seconds", 10, "Leader renewal frequency")

	flag.Parse()

	verbosity := 3 // TODO make this configurable 1-5 (5 is a security risk)
	opts := zap.Options{
		Level: zapcore.Level(-verbosity),
		EncoderConfigOptions: []zap.EncoderConfigOption{
			func(ec *zapcore.EncoderConfig) {
				ec.EncodeLevel = func(level zapcore.Level, enc zapcore.PrimitiveArrayEncoder) {
					enc.AppendInt8(-int8(level))
				}
			},
		},
	}
	zapLogger := zap.New(zap.UseFlagOptions(&opts))
	ctrl.SetLogger(zapLogger)
	klog.SetLogger(zapLogger)

	logger := ctrl.Log.WithName("portal_manager")
	setupLogger := logger.WithName("setup")

	var err error

	var ctrlConfig storageosv1alpha1.PortalConfig
	if configFile != "" {
		configContent, err := os.ReadFile(filepath.Clean(configFile))
		if err != nil {
			setupLogger.Error(err, "failed to read config file")
			panic(err)
		}

		switch path.Ext(configFile) {
		case ".yaml", ".yml":
			err = yaml.Unmarshal(configContent, &ctrlConfig)
			if err != nil {
				setupLogger.Error(err, "failed to parse config file")
				panic(err)
			}
		case ".json":
			err = json.Unmarshal(configContent, &ctrlConfig)
			if err != nil {
				setupLogger.Error(err, "failed to parse config file")
				panic(err)
			}
		default:
			err = errors.New("unsupported config format")
			setupLogger.Error(err, "unsupported config format, only YAML and JSON are supported")
			panic(err)
		}
	}

	currentNS := os.Getenv(podNamespace)
	if currentNS == "" {
		err = errors.New("current namespace not found")
		setupLogger.Error(err, "failed to get current namespace")
		panic(err)
	}

	if ctrlConfig.HTTPSProxy != "" {
		if err := os.Setenv("HTTPS_PROXY", ctrlConfig.HTTPSProxy); err != nil {
			setupLogger.Error(err, "failed to set environment variable")
			panic(err)
		}
	}

	restConfig := ctrl.GetConfigOrDie()

	kubeClient, err := client.New(restConfig, client.Options{})
	if err != nil {
		setupLogger.Error(err, "unable to build kubernetes client")
		panic(err)
	}

	// Setup StorageOS endpoint.
	stosUsername, err := utils.ReadFile(path.Join(stosSecretPath, stosUsernameKey))
	if err != nil {
		setupLogger.Error(err, "unable to read username")
		panic(err)
	}
	stosPassword, err := utils.ReadFile(path.Join(stosSecretPath, stosPasswordKey))
	if err != nil {
		setupLogger.Error(err, "unable to read password")
		panic(err)
	}
	stosService := fmt.Sprintf(stosServiceTemplate, currentNS)

	stosEndpoint := endpoints.NewStorageOSEndpoint(stosUsername, stosPassword, stosService, logger)
	err = stosEndpoint.Start(ctx)
	if err != nil {
		setupLogger.Error(err, "unable to start storageos endpoint")
		panic(err)
	}

	// Setup Portal API.
	portalURL, ok := os.LookupEnv(portalURLEnv)
	if !ok {
		setupLogger.Error(errors.New("missing key"), portalURLEnv)
		panic(err)
	}
	portalClientID, ok := os.LookupEnv(clientIDEnv)
	if !ok {
		setupLogger.Error(errors.New("missing key"), clientIDEnv)
		panic(err)
	}
	portalPassword, ok := os.LookupEnv(passwordEnv)
	if !ok {
		setupLogger.Error(errors.New("missing key"), passwordEnv)
		panic(err)
	}
	portalEndpoint := endpoints.NewPortalEndpoint(stosEndpoint.GetClusterID(), portalURL, portalClientID, portalPassword, logger)
	portalEndpoint.Start(ctx)

	// Setup device manager.
	deviceManager := managers.NewDeviceManager(kubeClient, *portalEndpoint, currentNS, logger)
	if err := deviceManager.Init(); err != nil {
		setupLogger.Error(err, "failed to start device manager")
		panic(err)
	}

	if ctrlConfig.IOTCore == nil {
		iotConfig, err := portalEndpoint.GetConfig()
		if err != nil {
			setupLogger.Error(err, "unable to retrieve config from portal")
			panic(err)
		}
		ctrlConfig.IOTCore = iotConfig
	}

	renewDeadline := time.Duration(leaderRenewSeconds) * time.Second
	leaseDuration := time.Duration(int(1.2*float64(leaderRenewSeconds))) * time.Second

	options := ctrl.Options{
		Scheme:                     scheme,
		LeaderElectionID:           "storageos-portal-manager-leader",
		LeaderElectionResourceLock: resourcelock.LeasesResourceLock,
		LeaderElectionNamespace:    currentNS,
		RenewDeadline:              &renewDeadline,
		LeaseDuration:              &leaseDuration,
		RetryPeriod:                &leaderRetryDuration,
	}
	if configFile != "" {
		options, err = options.AndFrom(ctrl.ConfigFile().AtPath(configFile).OfKind(&ctrlConfig))
		if err != nil {
			setupLogger.Error(err, "unable to load the config file")
			panic(err)
		}
	}

	// Setup controller manager.
	mgr, err := ctrl.NewManager(restConfig, options)
	if err != nil {
		setupLogger.Error(err, "unable to start manager")
		panic(err)
	}

	publishQueue := workqueue.NewRateLimitingQueue(workqueue.NewItemExponentialFailureRateLimiter(time.Second, time.Minute))

	tenantID, ok := os.LookupEnv(tenantIDEnv)
	if !ok {
		setupLogger.Error(errors.New("missing key"), tenantIDEnv)
		panic(err)
	}

	// Start action service.
	actionService := action.NewActionService(logger, stosEndpoint.GetClusterID(), []byte(ctrlConfig.IOTCore.SignatureKey), licence.NewLicenceActionHandler(*stosEndpoint, logger))

	// Start message broker.
	sink, err := publisher.New(stosEndpoint.GetClusterID(), deviceManager.GetPrivateKeyPem(), ctrlConfig, actionService.Do, logger)
	if err != nil {
		setupLogger.Error(err, "unable to create sink")
		panic(err)
	}
	if err := sink.Start(ctx); err != nil {
		setupLogger.Error(err, "unable to initialize sink")
		panic(err)
	}

	// Send acceptsConfiguration.
	go func() {
		ctx, cancel := context.WithTimeout(ctx, time.Minute)
		defer cancel()

		if err := sink.PublishState(ctx, &map[string]string{"acceptsConfiguration": "true"}); err != nil {
			setupLogger.Error(err, "unable to send initial state")
			panic(err)
		}

		currentLicence, err := stosEndpoint.GetLicence()
		if err != nil {
			setupLogger.Error(err, "unable to fetch licence")
			panic(err)
		}

		licenceState := state.ActionState{
			Meta: &state.ActionMeta{
				ClusterID:    stosEndpoint.GetClusterID(),
				Configurable: true,
				Success:      true,
			},
			Licence: &state.Licence{
				ExpireDate:  currentLicence.ExpiresAt,
				LicenceType: currentLicence.Kind,
			},
		}

		if err := sink.PublishState(ctx, &licenceState); err != nil {
			setupLogger.Error(err, "unable to send licence state")
			panic(err)
		}
	}()

	// Start config change watcher.
	configController := controllers.ConfigReconciler{
		Client: mgr.GetClient(),
		Scheme: mgr.GetScheme(),
	}
	if err = configController.SetupWithManager(mgr, currentNS); err != nil {
		setupLogger.Error(err, "unable to initialize controller", "controller", "config")
		panic(err)
	}

	// Start publisher.
	publisherService, err := controllers.NewPublisher(tenantID, stosEndpoint.GetClusterID(), sink, mgr.GetScheme(), publishQueue, logger)
	if err != nil {
		setupLogger.Error(err, "unable to create controller", "controller", "publisher")
		panic(err)
	}
	if err = publisherService.SetupWithManager(mgr); err != nil {
		setupLogger.Error(err, "unable to initialize controller", "controller", "publisher")
		panic(err)
	}

	// TODO(sc): WatchTypes need to be configurable via the PortalConfig CR.
	if err = (&controllers.WatchReconciler{
		Client: mgr.GetClient(),
		Scheme: mgr.GetScheme(),
		WatchTypes: []client.Object{
			&corev1.Node{},
			&corev1.Namespace{},
			&corev1.PersistentVolumeClaim{},
			&corev1.PersistentVolume{},
			&corev1.Pod{},
			&storagev1.StorageClass{},
			&storagev1.VolumeAttachment{},
			&appsv1.StatefulSet{},
			&appsv1.ReplicaSet{},
			&appsv1.Deployment{},
			&appsv1.DaemonSet{},
			&stosv1.Node{},
			&stosv1.Volume{},
			&stosoperatorv1.StorageOSCluster{},
		},
		EventHandler: handler.NewEventQueuer(publishQueue, logger),
	}).SetupWithManager(mgr); err != nil {
		setupLogger.Error(err, "unable to create controller", "controller", "portal")
		panic(err)
	}
	//+kubebuilder:scaffold:builder

	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
		setupLogger.Error(err, "unable to set up health check")
		panic(err)
	}
	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
		setupLogger.Error(err, "unable to set up ready check")
		panic(err)
	}

	setupLogger.V(3).Info("Starting manager...")
	if err := mgr.Start(ctx); err != nil {
		setupLogger.Error(err, "problem running manager")
		panic(err)
	}
}