feat: poc for flagd as a crd proxy

cmd/start.go → cmd/provider.go

@@ -7,6 +7,7 @@ import (
 	"github.com/open-feature/flagd/pkg/logger"
 	"github.com/open-feature/flagd/pkg/runtime"
 	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -27,9 +28,21 @@ const (
 	uriFlagName            = "uri"
 )
 
-func init() {
-	flags := startCmd.Flags()
+// NewProviderCmd is the command to start flagd as a provider
+func NewProviderCmd() *cobra.Command {
+	flagd := &cobra.Command{
+		Use:   "start",
+		Short: "Start flagd",
+		Long:  ``,
+		Run:   runProvider,
+	}
 
+	setupProvider(flagd.Flags())
+	return flagd
+}
+
+// setupProvider setup flags of the command
+func setupProvider(flags *pflag.FlagSet) {
 	// allows environment variables to use _ instead of -
 	viper.SetEnvKeyReplacer(strings.NewReplacer("-", "_")) // sync-provider-args becomes SYNC_PROVIDER_ARGS
 	viper.SetEnvPrefix("FLAGD")                            // port becomes FLAGD_PORT
@@ -72,54 +85,49 @@ func init() {
 	_ = viper.BindPFlag(uriFlagName, flags.Lookup(uriFlagName))
 }
 
-// startCmd represents the start command
-var startCmd = &cobra.Command{
-	Use:   "start",
-	Short: "Start flagd",
-	Long:  ``,
-	Run: func(cmd *cobra.Command, args []string) {
-		// Configure loggers -------------------------------------------------------
-		var level zapcore.Level
-		var err error
-		if Debug {
-			level = zapcore.DebugLevel
-		} else {
-			level = zapcore.InfoLevel
-		}
-		l, err := logger.NewZapLogger(level, viper.GetString(logFormatFlagName))
-		if err != nil {
-			log.Fatalf("can't initialize zap logger: %v", err)
-		}
-		logger := logger.NewLogger(l, Debug)
-		rtLogger := logger.WithFields(zap.String("component", "start"))
+// runProvider starts the provider implementation
+func runProvider(cmd *cobra.Command, args []string) {
+	// Configure loggers -------------------------------------------------------
+	var level zapcore.Level
+	var err error
+	if Debug {
+		level = zapcore.DebugLevel
+	} else {
+		level = zapcore.InfoLevel
+	}
+	l, err := logger.NewZapLogger(level, viper.GetString(logFormatFlagName))
+	if err != nil {
+		log.Fatalf("can't initialize zap logger: %v", err)
+	}
+	logger := logger.NewLogger(l, Debug)
+	rtLogger := logger.WithFields(zap.String("component", "start"))
 
-		if viper.GetString(syncProviderFlagName) != "" {
-			rtLogger.Warn("DEPRECATED: The --sync-provider flag has been deprecated. " +
-				"Docs: https://github.com/open-feature/flagd/blob/main/docs/configuration/configuration.md")
-		}
+	if viper.GetString(syncProviderFlagName) != "" {
+		rtLogger.Warn("DEPRECATED: The --sync-provider flag has been deprecated. " +
+			"Docs: https://github.com/open-feature/flagd/blob/main/docs/configuration/configuration.md")
+	}
 
-		if viper.GetString(evaluatorFlagName) != "json" {
-			rtLogger.Warn("DEPRECATED: The --evaluator flag has been deprecated. " +
-				"Docs: https://github.com/open-feature/flagd/blob/main/docs/configuration/configuration.md")
-		}
-		// Build Runtime -----------------------------------------------------------
-		rt, err := runtime.FromConfig(logger, runtime.Config{
-			CORS:              viper.GetStringSlice(corsFlagName),
-			MetricsPort:       viper.GetInt32(metricsPortFlagName),
-			ProviderArgs:      viper.GetStringMapString(providerArgsFlagName),
-			ServiceCertPath:   viper.GetString(serverCertPathFlagName),
-			ServiceKeyPath:    viper.GetString(serverKeyPathFlagName),
-			ServicePort:       viper.GetInt32(portFlagName),
-			ServiceSocketPath: viper.GetString(socketPathFlagName),
-			SyncBearerToken:   viper.GetString(bearerTokenFlagName),
-			SyncURI:           viper.GetStringSlice(uriFlagName),
-		})
-		if err != nil {
-			rtLogger.Fatal(err.Error())
-		}
+	if viper.GetString(evaluatorFlagName) != "json" {
+		rtLogger.Warn("DEPRECATED: The --evaluator flag has been deprecated. " +
+			"Docs: https://github.com/open-feature/flagd/blob/main/docs/configuration/configuration.md")
+	}
+	// Build Runtime -----------------------------------------------------------
+	rt, err := runtime.FromConfig(logger, runtime.Config{
+		CORS:              viper.GetStringSlice(corsFlagName),
+		MetricsPort:       viper.GetInt32(metricsPortFlagName),
+		ProviderArgs:      viper.GetStringMapString(providerArgsFlagName),
+		ServiceCertPath:   viper.GetString(serverCertPathFlagName),
+		ServiceKeyPath:    viper.GetString(serverKeyPathFlagName),
+		ServicePort:       viper.GetInt32(portFlagName),
+		ServiceSocketPath: viper.GetString(socketPathFlagName),
+		SyncBearerToken:   viper.GetString(bearerTokenFlagName),
+		SyncURI:           viper.GetStringSlice(uriFlagName),
+	})
+	if err != nil {
+		rtLogger.Fatal(err.Error())
+	}
 
-		if err := rt.Start(); err != nil {
-			rtLogger.Fatal(err.Error())
-		}
-	},
+	if err := rt.Start(); err != nil {
+		rtLogger.Fatal(err.Error())
+	}
 }

cmd/root.go

@@ -48,7 +48,8 @@ func init() {
 	// will be global for your application.
 	rootCmd.PersistentFlags().BoolVarP(&Debug, "debug", "x", false, "verbose logging")
 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file (default is $HOME/.agent.yaml)")
-	rootCmd.AddCommand(startCmd)
+	rootCmd.AddCommand(NewProviderCmd())
+	rootCmd.AddCommand(NewServerCmd())
 	rootCmd.AddCommand(versionCmd)
 }
 

cmd/server.go

@@ -0,0 +1,86 @@
+package cmd
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/open-feature/flagd/pkg/logger"
+	"github.com/open-feature/flagd/pkg/runtime"
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
+	"go.uber.org/zap/zapcore"
+)
+
+const (
+	address  = "address"
+	secure   = "secure"
+	certPath = "cert-path"
+	keyPath  = "key-path"
+	source   = "source"
+)
+
+// NewServerCmd is the command to start flagd in server mode
+func NewServerCmd() *cobra.Command {
+	flagdCmd := &cobra.Command{
+		Use:   "server",
+		Short: "Start flagd as a server",
+		Run:   runServer,
+	}
+
+	setupServer(flagdCmd)
+	return flagdCmd
+}
+
+// setupServer setup flags of the command
+func setupServer(cmd *cobra.Command) {
+	flags := cmd.Flags()
+
+	flags.StringP(address, "p", "localhost:9090", "Path this server binds to")
+
+	flags.BoolP(secure, "s", false, "Start secure server")
+	flags.StringP(certPath, "c", "", "TLS certificate path")
+	flags.StringP(keyPath, "k", "", "TLS key path of the certificate")
+	cmd.MarkFlagsRequiredTogether(secure, certPath, keyPath)
+
+	flags.StringP(source, "f", "", "CRD with feature flag configurations")
+
+	_ = viper.BindPFlag(address, flags.Lookup(address))
+	_ = viper.BindPFlag(secure, flags.Lookup(secure))
+	_ = viper.BindPFlag(certPath, flags.Lookup(certPath))
+	_ = viper.BindPFlag(keyPath, flags.Lookup(keyPath))
+	_ = viper.BindPFlag(source, flags.Lookup(source))
+}
+
+func runServer(cmd *cobra.Command, args []string) {
+	// todo align log format with provider runtime
+	zapLogger, err := logger.NewZapLogger(zapcore.DebugLevel, "console")
+	if err != nil {
+		log.Fatalf("error setting up the logger: %s", err)
+	}
+
+	logWrapper := logger.NewLogger(zapLogger, true)
+
+	err = viper.BindPFlags(pflag.CommandLine)
+	if err != nil {
+		logWrapper.Fatal(fmt.Sprintf("error parsing flags: %s", err.Error()))
+	}
+
+	serverConfig := runtime.ServerConfig{
+		Address:     viper.GetString(address),
+		Secure:      viper.GetBool(secure),
+		CertPath:    viper.GetString(certPath),
+		KeyPath:     viper.GetString(keyPath),
+		SyncSources: viper.GetString(source),
+	}
+
+	serverRuntime, err := runtime.NewServerRuntime(serverConfig, logWrapper)
+	if err != nil {
+		logWrapper.Fatal(fmt.Sprintf("error creating the server runtime: %s", err.Error()))
+	}
+
+	err = serverRuntime.Start()
+	if err != nil {
+		logWrapper.Fatal(fmt.Sprintf("error from server runtime: %s", err.Error()))
+	}
+}

docs/configuration/flagd.md

@@ -12,6 +12,7 @@ Flagd is a simple command line tool for fetching and presenting feature flags to
 
 ### SEE ALSO
 
+* [flagd server](flagd_server.md)	 - Start flagd as a server
 * [flagd start](flagd_start.md)	 - Start flagd
 * [flagd version](flagd_version.md)	 - Print the version number of FlagD
 

docs/configuration/flagd_server.md

@@ -0,0 +1,30 @@
+## flagd server
+
+Start flagd as a server
+
+```
+flagd server [flags]
+```
+
+### Options
+
+```
+  -p, --address string     Path this server binds to (default "localhost:9090")
+  -c, --cert-path string   TLS certificate path
+  -h, --help               help for server
+  -k, --key-path string    TLS key path of the certificate
+  -s, --secure             Start secure server
+  -f, --source string      CRD with feature flag configurations
+```
+
+### Options inherited from parent commands
+
+```
+      --config string   config file (default is $HOME/.agent.yaml)
+  -x, --debug           verbose logging
+```
+
+### SEE ALSO
+
+* [flagd](flagd.md)	 - Flagd is a simple command line tool for fetching and presenting feature flags to services. It is designed to conform to Open Feature schema for flag definitions.
+

pkg/runtime/serverRuntime.go

@@ -0,0 +1,99 @@
+package runtime
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/open-feature/flagd/pkg/server"
+	"github.com/open-feature/flagd/pkg/sync/kubernetes"
+	"go.uber.org/zap"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/open-feature/flagd/pkg/logger"
+	"github.com/open-feature/flagd/pkg/sync"
+)
+
+type ServerConfig struct {
+	Address     string
+	Secure      bool
+	CertPath    string
+	KeyPath     string
+	SyncSources string
+}
+
+type ServerRuntime struct {
+	syncProvider sync.ISync
+	logger       *logger.Logger
+	config       ServerConfig
+}
+
+func NewServerRuntime(config ServerConfig, rootLogger *logger.Logger) (*ServerRuntime, error) {
+	syncImpl, err := buildSyncImpl(config.SyncSources, rootLogger)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ServerRuntime{
+		syncProvider: syncImpl,
+		logger:       rootLogger.WithFields(zap.String("component", "Server Runtime")),
+		config:       config,
+	}, nil
+}
+
+func (sr *ServerRuntime) Start() error {
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+
+	// Build server
+	s := server.Server{
+		Logger:   sr.logger.WithFields(zap.String("component", "Server")),
+		Secure:   sr.config.Secure,
+		CertPath: sr.config.CertPath,
+		KeyPath:  sr.config.KeyPath,
+		Address:  sr.config.Address,
+	}
+
+	g, gCtx := errgroup.WithContext(ctx)
+	dataSync := make(chan sync.DataSync)
+
+	// Start server
+	g.Go(func() error {
+		return s.Listen(gCtx, dataSync)
+	})
+
+	// Start sync provider
+	g.Go(func() error {
+		return sr.syncProvider.Sync(gCtx, dataSync)
+	})
+
+	if err := g.Wait(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func buildSyncImpl(source string, rootLogger *logger.Logger) (sync.ISync, error) {
+	if len(source) == 0 {
+		return nil, errors.New("no sync provider sources provided")
+	}
+
+	switch sourceBytes := []byte(source); {
+	case regCrd.Match(sourceBytes):
+		rootLogger.Debug(fmt.Sprintf("using kubernetes sync-provider for: %s", source))
+		return &kubernetes.Sync{
+			Logger: rootLogger.WithFields(
+				zap.String("component", "sync"),
+				zap.String("sync", "kubernetes"),
+			),
+			URI: regCrd.ReplaceAllString(source, ""),
+		}, nil
+	default:
+		return nil, fmt.Errorf("server supports only crd sync provider, but received : %s", source)
+	}
+}