penguin/monok8s

clitools/pkg/cmd/controller/controller.go

raw · 3707 bytes

package controller

import (
	"context"
	"errors"
	"net"
	"net/http"
	"os"
	"time"

	"github.com/spf13/cobra"
	"k8s.io/cli-runtime/pkg/genericclioptions"
	"k8s.io/klog/v2"

	mkscontroller "example.com/monok8s/pkg/controller"
	osupgradectrl "example.com/monok8s/pkg/controller/osupgrade"
	"example.com/monok8s/pkg/kube"
	"example.com/monok8s/pkg/templates"
)

type ServerConfig struct {
	Namespace         string `json:"namespace,omitempty"`
	TLSCertFile       string `json:"tlsCertFile,omitempty"`
	TLSPrivateKeyFile string `json:"tlsPrivateKeyFile,omitempty"`
}

func NewCmdController(flags *genericclioptions.ConfigFlags) *cobra.Command {
	var conf ServerConfig

	cmd := &cobra.Command{
		Use:   "controller",
		Short: "Start a controller that handles OSUpgrade resources",
		RunE: func(cmd *cobra.Command, _ []string) error {
			ctx := cmd.Context()

			klog.InfoS("starting controller", "namespace", conf.Namespace)

			clients, err := kube.NewClients(flags)
			if err != nil {
				return err
			}

			ctx, cancel := context.WithCancel(ctx)
			defer cancel()

			httpErrCh := make(chan error, 1)
			watchErrCh := make(chan error, 1)

			go func() {
				klog.InfoS("starting OSUpgrade watch loop", "namespace", conf.Namespace)
				watchErrCh <- osupgradectrl.Watch(ctx, clients, conf.Namespace)
			}()

			go func() {
				httpErrCh <- httpListen(ctx, clients, conf)
			}()

			select {
			case <-ctx.Done():
				klog.InfoS("controller context canceled")
				return ctx.Err()

			case err := <-watchErrCh:
				if err != nil && !errors.Is(err, context.Canceled) {
					cancel()
					return err
				}
				cancel()
				return nil

			case err := <-httpErrCh:
				if err != nil && !errors.Is(err, context.Canceled) {
					cancel()
					return err
				}
				cancel()
				return nil
			}
		},
	}

	cmd.Flags().StringVar(&conf.Namespace, "namespace", templates.DefaultNamespace, "namespace to watch")
	cmd.Flags().StringVar(&conf.TLSCertFile, "tls-cert-file", conf.TLSCertFile,
		"File containing x509 Certificate used for serving HTTPS (with intermediate certs, if any, concatenated after server cert).")
	cmd.Flags().StringVar(&conf.TLSPrivateKeyFile, "tls-private-key-file", conf.TLSPrivateKeyFile,
		"File containing x509 private key matching --tls-cert-file.")

	return cmd
}

func httpListen(ctx context.Context, clients *kube.Clients, conf ServerConfig) error {
	address, port := "", "8443"
	addr := net.JoinHostPort(address, port)

	nodeName := os.Getenv("NODE_NAME")

	server := mkscontroller.NewServer(ctx, clients, conf.Namespace, nodeName)

	s := &http.Server{
		Addr:           addr,
		Handler:        server,
		IdleTimeout:    90 * time.Second,
		ReadTimeout:    4 * time.Minute,
		WriteTimeout:   4 * time.Minute,
		MaxHeaderBytes: 1 << 20,
	}

	serverErrCh := make(chan error, 1)

	go func() {
		if conf.TLSCertFile != "" {
			klog.InfoS("starting HTTPS server",
				"addr", addr,
				"certFile", conf.TLSCertFile,
				"keyFile", conf.TLSPrivateKeyFile,
			)
			serverErrCh <- s.ListenAndServeTLS(conf.TLSCertFile, conf.TLSPrivateKeyFile)
			return
		}

		klog.InfoS("starting HTTP server", "addr", addr)
		serverErrCh <- s.ListenAndServe()
	}()

	select {
	case <-ctx.Done():
		klog.InfoS("shutting down HTTP server", "addr", addr)

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

		err := s.Shutdown(shutdownCtx)
		if err != nil {
			return err
		}

		err = <-serverErrCh
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			return err
		}
		return context.Canceled

	case err := <-serverErrCh:
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			klog.ErrorS(err, "HTTP server failed")
			return err
		}
		return nil
	}
}