Add resource version for status reporting and update configuration types

cmd/botkube/main.go

@@ -27,6 +27,7 @@ import (
 	"github.com/kubeshop/botkube/internal/analytics"
 	"github.com/kubeshop/botkube/internal/audit"
 	"github.com/kubeshop/botkube/internal/command"
+	intconfig "github.com/kubeshop/botkube/internal/config"
 	"github.com/kubeshop/botkube/internal/graphql"
 	"github.com/kubeshop/botkube/internal/lifecycle"
 	"github.com/kubeshop/botkube/internal/loggerx"
@@ -48,12 +49,13 @@ import (
 )
 
 const (
-	componentLogFieldKey = "component"
-	botLogFieldKey       = "bot"
-	sinkLogFieldKey      = "sink"
-	commGroupFieldKey    = "commGroup"
-	healthEndpointName   = "/healthz"
-	printAPIKeyCharCount = 3
+	componentLogFieldKey  = "component"
+	botLogFieldKey        = "bot"
+	sinkLogFieldKey       = "sink"
+	commGroupFieldKey     = "commGroup"
+	healthEndpointName    = "/healthz"
+	printAPIKeyCharCount  = 3
+	configUpdaterInterval = 15 * time.Second
 )
 
 func main() {
@@ -70,10 +72,14 @@ func main() {
 // run wraps the main logic of the app to be able to properly clean up resources via deferred calls.
 func run(ctx context.Context) error {
 	// Load configuration
-	config.RegisterFlags(pflag.CommandLine)
+	intconfig.RegisterFlags(pflag.CommandLine)
+
+	remoteCfgSyncEnabled := graphql.IsRemoteConfigEnabled()
 	gqlClient := graphql.NewDefaultGqlClient()
-	cfgProvider := config.GetProvider(gqlClient)
-	configs, err := cfgProvider.Configs(ctx)
+	deployClient := intconfig.NewDeploymentClient(gqlClient)
+
+	cfgProvider := intconfig.GetProvider(remoteCfgSyncEnabled, deployClient)
+	configs, cfgVersion, err := cfgProvider.Configs(ctx)
 	if err != nil {
 		return fmt.Errorf("while loading configuration files: %w", err)
 	}
@@ -84,13 +90,13 @@ func run(ctx context.Context) error {
 	}
 
 	logger := loggerx.New(conf.Settings.Log)
-	statusReporter := status.NewStatusReporter(logger, gqlClient)
-	auditReporter := audit.NewAuditReporter(logger, gqlClient)
-
 	if confDetails.ValidateWarnings != nil {
 		logger.Warnf("Configuration validation warnings: %v", confDetails.ValidateWarnings.Error())
 	}
 
+	statusReporter := status.NewStatusReporter(remoteCfgSyncEnabled, logger, gqlClient, deployClient, cfgVersion)
+	auditReporter := audit.NewAuditReporter(remoteCfgSyncEnabled, logger, gqlClient)
+
 	// Set up analytics reporter
 	reporter, err := newAnalyticsReporter(conf.Analytics.Disable, logger)
 	if err != nil {
@@ -302,6 +308,18 @@ func run(ctx context.Context) error {
 		})
 	}
 
+	cfgUpdater := intconfig.GetConfigUpdater(
+		remoteCfgSyncEnabled,
+		logger.WithField(componentLogFieldKey, "Config Updater"),
+		configUpdaterInterval,
+		deployClient,
+		statusReporter,
+	)
+	errGroup.Go(func() error {
+		defer analytics.ReportPanicIfOccurs(logger, reporter)
+		return cfgUpdater.Do(ctx)
+	})
+
 	if conf.ConfigWatcher.Enabled {
 		err := config.WaitForWatcherSync(
 			ctx,
@@ -358,7 +376,7 @@ func run(ctx context.Context) error {
 		statusReporter,
 	)
 
-	if _, err := statusReporter.ReportDeploymentStartup(ctx); err != nil {
+	if err := statusReporter.ReportDeploymentStartup(ctx); err != nil {
 		return reportFatalError("while reporting botkube startup", err)
 	}
 
@@ -462,8 +480,8 @@ func reportFatalErrFn(logger logrus.FieldLogger, reporter analytics.Reporter, st
 			logger.Errorf("while reporting fatal error: %s", err.Error())
 		}
 
-		if _, err := status.ReportDeploymentFailed(ctxTimeout); err != nil {
-			logger.Errorf("while reporting botkube deployment status: %s", err.Error())
+		if err := status.ReportDeploymentFailed(ctxTimeout); err != nil {
+			logger.Errorf("while reporting deployment failure: %s", err.Error())
 		}
 
 		return wrappedErr

go.mod

@@ -3,6 +3,7 @@ module github.com/kubeshop/botkube
 require (
 	github.com/MakeNowJust/heredoc v1.0.0
 	github.com/alexflint/go-arg v1.4.3
+	github.com/avast/retry-go v3.0.0+incompatible
 	github.com/aws/aws-sdk-go v1.44.122
 	github.com/bwmarrin/discordgo v0.25.0
 	github.com/dustin/go-humanize v1.0.0
@@ -51,7 +52,6 @@ require (
 	k8s.io/kubectl v0.25.4
 	k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2
 	sigs.k8s.io/controller-runtime v0.13.1
-	sigs.k8s.io/yaml v1.3.0
 )
 
 require (
@@ -196,6 +196,7 @@ require (
 	sigs.k8s.io/kustomize/api v0.12.1 // indirect
 	sigs.k8s.io/kustomize/kyaml v0.13.9 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
+	sigs.k8s.io/yaml v1.3.0 // indirect
 )
 
 go 1.19

go.sum

@@ -328,6 +328,8 @@ github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj
 github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
+github.com/avast/retry-go v3.0.0+incompatible h1:4SOWQ7Qs+oroOTQOYnAHqelpCO0biHSxpiH9JdtuBj0=
+github.com/avast/retry-go v3.0.0+incompatible/go.mod h1:XtSnn+n/sHqQIpZ10K1qAevBhOOCWBLXXy3hyiqqBrY=
 github.com/avct/uasurfer v0.0.0-20191028135549-26b5daa857f1/go.mod h1:noBAuukeYOXa0aXGqxr24tADqkwDO2KRD15FsuaZ5a8=
 github.com/aws/aws-sdk-go v1.15.11/go.mod h1:mFuSZ37Z9YOHbQEwBWztmVzqXrEkub65tZoCYDt7FT0=
 github.com/aws/aws-sdk-go v1.17.7/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=

internal/audit/reporter.go

@@ -2,7 +2,6 @@ package audit
 
 import (
 	"context"
-	"os"
 
 	"github.com/sirupsen/logrus"
 
@@ -34,8 +33,8 @@ type SourceAuditEvent struct {
 }
 
 // NewAuditReporter creates new AuditReporter
-func NewAuditReporter(logger logrus.FieldLogger, gql *graphql.Gql) AuditReporter {
-	if _, provided := os.LookupEnv(graphql.GqlProviderIdentifierEnvKey); provided {
+func NewAuditReporter(remoteCfgSyncEnabled bool, logger logrus.FieldLogger, gql *graphql.Gql) AuditReporter {
+	if remoteCfgSyncEnabled {
 		return newGraphQLAuditReporter(logger.WithField("component", "GraphQLAuditReporter"), gql)
 	}
 	return newNoopAuditReporter(nil)

internal/config/env_provider.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"os"
 	"strings"
+
+	"github.com/kubeshop/botkube/pkg/config"
 )
 
 const (
@@ -20,7 +22,7 @@ func NewEnvProvider() *EnvProvider {
 }
 
 // Configs returns list of config file locations
-func (e *EnvProvider) Configs(ctx context.Context) (YAMLFiles, error) {
+func (e *EnvProvider) Configs(ctx context.Context) (config.YAMLFiles, int, error) {
 	envCfgs := os.Getenv(EnvProviderConfigPathsEnvKey)
 	configPaths := strings.Split(envCfgs, ",")
 

internal/config/env_provider_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestEnvProviderSuccess(t *testing.T) {
@@ -14,19 +15,20 @@ func TestEnvProviderSuccess(t *testing.T) {
 
 	// when
 	p := NewEnvProvider()
-	configs, err := p.Configs(context.Background())
+	configs, cfgVer, err := p.Configs(context.Background())
 
 	// then
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	content, err := os.ReadFile("testdata/TestEnvProviderSuccess/config.yaml")
 	assert.NoError(t, err)
 	assert.Equal(t, content, configs[0])
+	assert.Equal(t, cfgVer, 0)
 }
 
 func TestEnvProviderErr(t *testing.T) {
 	// when
 	p := NewEnvProvider()
-	_, err := p.Configs(context.Background())
+	_, _, err := p.Configs(context.Background())
 
 	// then
 	assert.Equal(t, "while reading a file: read .: is a directory", err.Error())

internal/config/flag.go

@@ -0,0 +1,10 @@
+package config
+
+import "github.com/spf13/pflag"
+
+var configPathsFlag []string
+
+// RegisterFlags registers config related flags.
+func RegisterFlags(flags *pflag.FlagSet) {
+	flags.StringSliceVarP(&configPathsFlag, "config", "c", nil, "Specify configuration file in YAML format (can specify multiple).")
+}

internal/config/fs_provider.go

@@ -6,6 +6,8 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+
+	"github.com/kubeshop/botkube/pkg/config"
 )
 
 const specialConfigFileNamePrefix = "_"
@@ -21,19 +23,19 @@ func NewFileSystemProvider(configs []string) *FileSystemProvider {
 }
 
 // Configs returns list of config file locations.
-func (e *FileSystemProvider) Configs(_ context.Context) (YAMLFiles, error) {
+func (e *FileSystemProvider) Configs(_ context.Context) (config.YAMLFiles, int, error) {
 	configPaths := sortCfgFiles(e.Files)
 
-	var out YAMLFiles
+	var out config.YAMLFiles
 	for _, path := range configPaths {
 		raw, err := os.ReadFile(filepath.Clean(path))
 		if err != nil {
-			return nil, fmt.Errorf("while reading a file: %w", err)
+			return nil, 0, fmt.Errorf("while reading a file: %w", err)
 		}
 		out = append(out, raw)
 	}
 
-	return out, nil
+	return out, 0, nil
 }
 
 // sortCfgFiles sorts the config files so that the files that has specialConfigFileNamePrefix are moved to the end of the slice.

internal/config/fs_provider_test.go

@@ -6,18 +6,20 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestStaticProviderSuccess(t *testing.T) {
 	// when
 	p := NewFileSystemProvider([]string{"testdata/TestStaticProviderSuccess/config.yaml"})
-	configs, err := p.Configs(context.Background())
+	configs, cfgVer, err := p.Configs(context.Background())
 
 	// then
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	content, err := os.ReadFile("testdata/TestStaticProviderSuccess/config.yaml")
 	assert.NoError(t, err)
 	assert.Equal(t, content, configs[0])
+	assert.Equal(t, cfgVer, 0)
 }
 
 func TestSortCfgFiles(t *testing.T) {

internal/config/gql_client.go

@@ -11,7 +11,7 @@ import (
 
 // DeploymentClient defines GraphQL client.
 type DeploymentClient interface {
-	GetDeployment(ctx context.Context) (Deployment, error)
+	GetConfigWithResourceVersion(ctx context.Context) (Deployment, error)
 }
 
 // Gql defines GraphQL client data structure.
@@ -27,11 +27,12 @@ func NewDeploymentClient(client *gql.Gql) *Gql {
 
 // Deployment returns deployment with Botkube configuration.
 type Deployment struct {
-	BotkubeConfig string
+	ResourceVersion int
+	YAMLConfig      string
 }
 
-// GetDeployment retrieves deployment by id.
-func (g *Gql) GetDeployment(ctx context.Context) (Deployment, error) {
+// GetConfigWithResourceVersion retrieves deployment by id.
+func (g *Gql) GetConfigWithResourceVersion(ctx context.Context) (Deployment, error) {
 	var query struct {
 		Deployment Deployment `graphql:"deployment(id: $id)"`
 	}
@@ -40,7 +41,24 @@ func (g *Gql) GetDeployment(ctx context.Context) (Deployment, error) {
 	}
 	err := g.client.Cli.Query(ctx, &query, variables)
 	if err != nil {
-		return Deployment{}, fmt.Errorf("while querying deployment details for %q: %w", g.deploymentID, err)
+		return Deployment{}, fmt.Errorf("while getting config with resource version for %q: %w", g.deploymentID, err)
 	}
 	return query.Deployment, nil
 }
+
+// GetResourceVersion retrieves resource version for Deployment.
+func (g *Gql) GetResourceVersion(ctx context.Context) (int, error) {
+	var query struct {
+		Deployment struct {
+			ResourceVersion int
+		} `graphql:"deployment(id: $id)"`
+	}
+	variables := map[string]interface{}{
+		"id": graphql.ID(g.deploymentID),
+	}
+	err := g.client.Cli.Query(ctx, &query, variables)
+	if err != nil {
+		return 0, fmt.Errorf("while querying deployment details for %q: %w", g.deploymentID, err)
+	}
+	return query.Deployment.ResourceVersion, nil
+}

internal/config/gql_client_test.go

@@ -16,7 +16,7 @@ import (
 )
 
 func TestGql_GetDeployment(t *testing.T) {
-	expectedBody := fmt.Sprintf(`{"query":"query ($id:ID!){deployment(id: $id){botkubeConfig}}","variables":{"id":"my-id"}}%s`, "\n")
+	expectedBody := fmt.Sprintf(`{"query":"query ($id:ID!){deployment(id: $id){resourceVersion,yamlConfig}}","variables":{"id":"my-id"}}%s`, "\n")
 	file, err := os.ReadFile("testdata/gql_get_deployment_success.json")
 	assert.NoError(t, err)
 	svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -49,7 +49,7 @@ func TestGql_GetDeployment(t *testing.T) {
 		graphql.WithDeploymentID("my-id"),
 	)
 	g := NewDeploymentClient(gqlClient)
-	deployment, err := g.GetDeployment(context.Background())
+	deployment, err := g.GetConfigWithResourceVersion(context.Background())
 	assert.NoError(t, err)
-	assert.NotNil(t, deployment.BotkubeConfig)
+	assert.NotNil(t, deployment.YAMLConfig)
 }

internal/config/gql_provider.go

@@ -4,7 +4,8 @@ import (
 	"context"
 
 	"github.com/pkg/errors"
-	"sigs.k8s.io/yaml"
+
+	"github.com/kubeshop/botkube/pkg/config"
 )
 
 // GqlProvider is GraphQL provider
@@ -18,17 +19,13 @@ func NewGqlProvider(dc DeploymentClient) *GqlProvider {
 }
 
 // Configs returns list of config files
-func (g *GqlProvider) Configs(ctx context.Context) (YAMLFiles, error) {
-	deployment, err := g.client.GetDeployment(ctx)
-	if err != nil {
-		return nil, errors.Wrapf(err, "while getting deployment")
-	}
-	conf, err := yaml.JSONToYAML([]byte(deployment.BotkubeConfig))
+func (g *GqlProvider) Configs(ctx context.Context) (config.YAMLFiles, int, error) {
+	deployment, err := g.client.GetConfigWithResourceVersion(ctx)
 	if err != nil {
-		return nil, errors.Wrapf(err, "while converting json to yaml for deployment")
+		return nil, 0, errors.Wrapf(err, "while getting deployment")
 	}
 
 	return [][]byte{
-		conf,
-	}, nil
+		[]byte(deployment.YAMLConfig),
+	}, deployment.ResourceVersion, nil
 }