feat: support azure blob sync

Signed-off-by: Matthew Wilson <54033231+wilson-matthew@users.noreply.github.com>

core/go.mod

@@ -49,6 +49,13 @@ require (
 	cloud.google.com/go/compute/metadata v0.5.0 // indirect
 	cloud.google.com/go/iam v1.1.13 // indirect
 	cloud.google.com/go/storage v1.43.0 // indirect
+	github.com/Azure/azure-sdk-for-go/sdk/azcore v1.14.0 // indirect
+	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.7.0 // indirect
+	github.com/Azure/azure-sdk-for-go/sdk/internal v1.10.0 // indirect
+	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.3.2 // indirect
+	github.com/Azure/go-autorest v14.2.0+incompatible // indirect
+	github.com/Azure/go-autorest/autorest/to v0.4.0 // indirect
+	github.com/AzureAD/microsoft-authentication-library-for-go v1.2.2 // indirect
 	github.com/barkimedes/go-deepcopy v0.0.0-20220514131651-17c30cfc62df // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
@@ -65,6 +72,7 @@ require (
 	github.com/go-openapi/jsonreference v0.21.0 // indirect
 	github.com/go-openapi/swag v0.23.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang-jwt/jwt/v5 v5.2.1 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
@@ -80,10 +88,12 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.7 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect

core/pkg/sync/blob/blob_sync.go

@@ -10,7 +10,8 @@ import (
 	"github.com/open-feature/flagd/core/pkg/logger"
 	"github.com/open-feature/flagd/core/pkg/sync"
 	"gocloud.dev/blob"
-	_ "gocloud.dev/blob/gcsblob" // needed to initialize GCS driver
+	_ "gocloud.dev/blob/azureblob" // needed to initialize Azure Blob Storage driver
+	_ "gocloud.dev/blob/gcsblob"   // needed to initialize GCS driver
 )
 
 type Sync struct {

core/pkg/sync/builder/syncbuilder.go

@@ -31,6 +31,7 @@ const (
 	syncProviderKubernetes = "kubernetes"
 	syncProviderHTTP       = "http"
 	syncProviderGcs        = "gcs"
+	syncProviderAzblob     = "azblob"
 )
 
 var (
@@ -40,6 +41,7 @@ var (
 	regGRPCSecure *regexp.Regexp
 	regFile       *regexp.Regexp
 	regGcs        *regexp.Regexp
+	regAzblob     *regexp.Regexp
 )
 
 func init() {
@@ -49,6 +51,7 @@ func init() {
 	regGRPCSecure = regexp.MustCompile("^" + grpc.PrefixSecure)
 	regFile = regexp.MustCompile("^file:")
 	regGcs = regexp.MustCompile("^gs://.+?/")
+	regAzblob = regexp.MustCompile("^azblob://.+?/")
 }
 
 type ISyncBuilder interface {
@@ -111,11 +114,14 @@ func (sb *SyncBuilder) syncFromConfig(sourceConfig sync.SourceConfig, logger *lo
 	case syncProviderGcs:
 		logger.Debug(fmt.Sprintf("using blob sync-provider with gcs driver for: %s", sourceConfig.URI))
 		return sb.newGcs(sourceConfig, logger), nil
+	case syncProviderAzblob:
+		logger.Debug(fmt.Sprintf("using blob sync-provider with azblob driver for: %s", sourceConfig.URI))
+		return sb.newAzblob(sourceConfig, logger)
 
 	default:
-		return nil, fmt.Errorf("invalid sync provider: %s, must be one of with '%s', '%s', '%s', %s', '%s' or '%s'",
+		return nil, fmt.Errorf("invalid sync provider: %s, must be one of with '%s', '%s', '%s', '%s', '%s', '%s', '%s' or '%s'",
 			sourceConfig.Provider, syncProviderFile, syncProviderFsNotify, syncProviderFileInfo,
-			syncProviderKubernetes, syncProviderHTTP, syncProviderKubernetes)
+			syncProviderKubernetes, syncProviderHTTP, syncProviderGrpc, syncProviderGcs, syncProviderAzblob)
 	}
 }
 
@@ -238,6 +244,43 @@ func (sb *SyncBuilder) newGcs(config sync.SourceConfig, logger *logger.Logger) *
 	}
 }
 
+func (sb *SyncBuilder) newAzblob(config sync.SourceConfig, logger *logger.Logger) (*blobSync.Sync, error) {
+	// Required to generate the azblob service URL
+	storageAccountName := os.Getenv("AZURE_STORAGE_ACCOUNT")
+	if storageAccountName == "" {
+		return nil, fmt.Errorf("environment variable AZURE_STORAGE_ACCOUNT not set or is blank")
+	}
+	if regexp.MustCompile(`\s`).MatchString(storageAccountName) {
+		return nil, fmt.Errorf("environment variable AZURE_STORAGE_ACCOUNT contains whitespace")
+	}
+
+	// Extract bucket uri and object name from the full URI:
+	// azblob://bucket/path/to/object results in azblob://bucket/ as bucketUri and
+	// path/to/object as an object name.
+	bucketURI := regAzblob.FindString(config.URI)
+	objectName := regAzblob.ReplaceAllString(config.URI, "")
+
+	// Defaults to 5 seconds if interval is not set.
+	var interval uint32 = 5
+	if config.Interval != 0 {
+		interval = config.Interval
+	}
+
+	return &blobSync.Sync{
+		Bucket: bucketURI,
+		Object: objectName,
+
+		BlobURLMux: blob.DefaultURLMux(),
+
+		Logger: logger.WithFields(
+			zap.String("component", "sync"),
+			zap.String("sync", "azblob"),
+		),
+		Interval: interval,
+		Cron:     cron.New(),
+	}, nil
+}
+
 type IK8sClientBuilder interface {
 	GetK8sClient() (dynamic.Interface, error)
 }

core/pkg/sync/builder/syncbuilder_test.go

@@ -197,6 +197,8 @@ func Test_SyncsFromFromConfig(t *testing.T) {
 		{
 			name: "combined",
 			injectFunc: func(builder *SyncBuilder) {
+				t.Setenv("AZURE_STORAGE_ACCOUNT", "myaccount")
+
 				ctrl := gomock.NewController(t)
 
 				mockClientBuilder := buildermock.NewMockIK8sClientBuilder(ctrl)
@@ -236,6 +238,10 @@ func Test_SyncsFromFromConfig(t *testing.T) {
 						URI:      "gs://bucket/path/to/file",
 						Provider: syncProviderGcs,
 					},
+					{
+						URI:      "azblob://bucket/path/to/file",
+						Provider: syncProviderAzblob,
+					},
 				},
 			},
 			wantSyncs: []sync.ISync{
@@ -245,6 +251,7 @@ func Test_SyncsFromFromConfig(t *testing.T) {
 				&file.Sync{},
 				&kubernetes.Sync{},
 				&blob.Sync{},
+				&blob.Sync{},
 			},
 			wantErr: false,
 		},
@@ -324,3 +331,90 @@ func Test_GcsConfig(t *testing.T) {
 		})
 	}
 }
+
+func Test_AzblobConfig(t *testing.T) {
+	lg := logger.NewLogger(nil, false)
+	defaultInterval := uint32(5)
+	tests := []struct {
+		name             string
+		uri              string
+		interval         uint32
+		storageAccount   string
+		expectedBucket   string
+		expectedObject   string
+		expectedInterval uint32
+		wantErr          bool
+	}{
+		{
+			name:             "simple path",
+			uri:              "azblob://bucket/path/to/object",
+			interval:         10,
+			storageAccount:   "myaccount",
+			expectedBucket:   "azblob://bucket/",
+			expectedObject:   "path/to/object",
+			expectedInterval: 10,
+			wantErr:          false,
+		},
+		{
+			name:             "default interval",
+			uri:              "azblob://bucket/path/to/object",
+			storageAccount:   "myaccount",
+			expectedBucket:   "azblob://bucket/",
+			expectedObject:   "path/to/object",
+			expectedInterval: defaultInterval,
+			wantErr:          false,
+		},
+		{
+			name:             "no object set", // Blob syncer will return error when fetching
+			uri:              "azblob://bucket/",
+			storageAccount:   "myaccount",
+			expectedBucket:   "azblob://bucket/",
+			expectedObject:   "",
+			expectedInterval: defaultInterval,
+			wantErr:          false,
+		},
+		{
+			name:             "malformed uri", // Blob syncer will return error when opening bucket
+			uri:              "malformed",
+			storageAccount:   "myaccount",
+			expectedBucket:   "",
+			expectedObject:   "malformed",
+			expectedInterval: defaultInterval,
+			wantErr:          false,
+		},
+		{
+			name:           "storage account not set", // Sync builder will fail and return error
+			uri:            "azblob://bucket/path/to/object",
+			storageAccount: "",
+			wantErr:        true,
+		},
+		{
+			name:           "storage account contains whitespace", // Sync builder will fail and return error
+			uri:            "azblob://bucket/path/to/object",
+			storageAccount: "my account",
+			wantErr:        true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Setenv("AZURE_STORAGE_ACCOUNT", tt.storageAccount)
+			azblobSync, err := NewSyncBuilder().newAzblob(sync.SourceConfig{
+				URI:      tt.uri,
+				Interval: tt.interval,
+			}, lg)
+
+			if (err != nil) != tt.wantErr {
+				t.Errorf("newAzblob() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+
+			if (err != nil) && (tt.wantErr == true) {
+				return
+			}
+
+			require.Equal(t, tt.expectedBucket, azblobSync.Bucket)
+			require.Equal(t, tt.expectedObject, azblobSync.Object)
+			require.Equal(t, int(tt.expectedInterval), int(azblobSync.Interval))
+		})
+	}
+}

core/pkg/sync/builder/utils.go

@@ -69,9 +69,14 @@ func ParseSyncProviderURIs(uris []string) ([]sync.SourceConfig, error) {
 				URI:      uri,
 				Provider: syncProviderGcs,
 			})
+		case regAzblob.Match(uriB):
+			syncProvidersParsed = append(syncProvidersParsed, sync.SourceConfig{
+				URI:      uri,
+				Provider: syncProviderAzblob,
+			})
 		default:
 			return syncProvidersParsed, fmt.Errorf("invalid sync uri argument: %s, must start with 'file:', "+
-				"'http(s)://', 'grpc(s)://', 'gs://' or 'core.openfeature.dev'", uri)
+				"'http(s)://', 'grpc(s)://', 'gs://', 'azblob://' or 'core.openfeature.dev'", uri)
 		}
 	}
 	return syncProvidersParsed, nil

core/pkg/sync/builder/utils_test.go

@@ -29,7 +29,8 @@ func TestParseSource(t *testing.T) {
 					{"uri":"http://test.com","provider":"http","bearerToken":":)"},
 					{"uri":"host:port","provider":"grpc"},
 					{"uri":"default/my-crd","provider":"kubernetes"},
-					{"uri":"gs://bucket-name/path/to/file","provider":"gcs"}
+					{"uri":"gs://bucket-name/path/to/file","provider":"gcs"},
+					{"uri":"azblob://bucket-name/path/to/file","provider":"azblob"}
 				]`,
 			expectErr: false,
 			out: []sync.SourceConfig{
@@ -54,6 +55,10 @@ func TestParseSource(t *testing.T) {
 					URI:      "gs://bucket-name/path/to/file",
 					Provider: syncProviderGcs,
 				},
+				{
+					URI:      "azblob://bucket-name/path/to/file",
+					Provider: syncProviderAzblob,
+				},
 			},
 		},
 		"multiple-syncs-with-options": {
@@ -188,6 +193,7 @@ func TestParseSyncProviderURIs(t *testing.T) {
 				"grpcs://secure-grpc",
 				"core.openfeature.dev/default/my-crd",
 				"gs://bucket-name/path/to/file",
+				"azblob://bucket-name/path/to/file",
 			},
 			expectErr: false,
 			out: []sync.SourceConfig{
@@ -217,6 +223,10 @@ func TestParseSyncProviderURIs(t *testing.T) {
 					URI:      "gs://bucket-name/path/to/file",
 					Provider: syncProviderGcs,
 				},
+				{
+					URI:      "azblob://bucket-name/path/to/file",
+					Provider: syncProviderAzblob,
+				},
 			},
 		},
 		"empty": {

docs/concepts/syncs.md

@@ -72,7 +72,7 @@ See [sync source](../reference/sync-configuration.md#source-configuration) confi
 
 ### GCS sync
 
-The GCS sync provider fetches flags from a GCS blob and periodically poll the GCS for the flag definition updates.
+The GCS sync provider fetches flags from a GCS blob and periodically polls the GCS for the flag definition updates.
 It uses [application default credentials](https://cloud.google.com/docs/authentication/application-default-credentials) if they
 are [configured](https://cloud.google.com/docs/authentication/provide-credentials-adc) to authorize the calls to GCS.
 
@@ -85,6 +85,22 @@ In this example, `gs://my-bucket/my-flags.json` is expected to be a valid GCS UR
 The polling interval can be configured.
 See [sync source](../reference/sync-configuration.md#source-configuration) configuration for details.
 
+### Azure Blob sync
+
+The Azure Blob sync provider fetches flags from an Azure Blob Storage blob and periodically polls the blob for the flag definition updates.
+It uses [environment variables](https://pkg.go.dev/gocloud.dev/blob/azureblob#hdr-URLs) to set the Storage Account name and to
+authorize the calls to Azure Blob Storage.
+
+```shell
+flagd start --uri azblob://my-container/my-flags.json
+```
+
+In this example, assuming the environment variable AZURE_STORAGE_ACCOUNT is set to `myaccount`, and other options are not set, the service URL will be:
+"https://myaccount.blob.core.windows.net/my-container/my-flags.json".
+This is expected be a valid service URL accessible by flagd (either by being public or together with environment variable credentials).
+The polling interval can be configured.
+See [sync source](../reference/sync-configuration.md#source-configuration) configuration for details.
+
 ## Merging
 
 Flagd can be configured to read from multiple sources at once, when this is the case flagd will merge all flag definition into a single

docs/reference/sync-configuration.md

@@ -8,7 +8,7 @@ See [syncs](../concepts/syncs.md) for a conceptual overview.
 
 ## URI patterns
 
-Any URI passed to flagd via the `--uri` (`-f`) flag must follow one of the 4 following patterns with prefixes to ensure that
+Any URI passed to flagd via the `--uri` (`-f`) flag must follow one of the 6 following patterns with prefixes to ensure that
 it is passed to the correct implementation:
 
 | Implied Sync Provider | Prefix                 | Example                               |
@@ -18,6 +18,7 @@ it is passed to the correct implementation:
 | `http`                | `http(s)://`           | `https://my-flags.com/flags`          |
 | `grpc`                | `grpc(s)://`           | `grpc://my-flags-server`              |
 | `gcs`                 | `gs://`                | `gs://my-bucket/my-flags.json`        |
+| `azblob`              | `azblob://`            | `azblob://my-container/my-flags.json` |
 
 ## Source Configuration
 
@@ -31,10 +32,10 @@ Alternatively, these configurations can be passed to flagd via config file, spec
 | Field       | Type               | Note                                                                                                                                                                                                             |
 | ----------- | ------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | uri         | required `string`  | Flag configuration source of the sync                                                                                                                                                                            |
-| provider    | required `string`  | Provider type - `file`, `fsnotify`, `fileinfo`, `kubernetes`, `http`, `grpc` or `gcs`                                                                                                                                                        |
+| provider    | required `string`  | Provider type - `file`, `fsnotify`, `fileinfo`, `kubernetes`, `http`, `grpc`, `gcs` or `azblob`                                                                                                                  |
 | authHeader  | optional `string`  | Used for http sync; set this to include the complete `Authorization` header value for any authentication scheme (e.g., "Bearer token_here", "Basic base64_credentials", etc.). Cannot be used with `bearerToken` |
 | bearerToken | optional `string`  | (Deprecated) Used for http sync; token gets appended to `Authorization` header with [bearer schema](https://www.rfc-editor.org/rfc/rfc6750#section-2.1). Cannot be used with `authHeader`                        |
-| interval    | optional `uint32`  | Used for http and gcs syncs; requests will be made at this interval. Defaults to 5 seconds.                                                                                                                               |
+| interval    | optional `uint32`  | Used for http, gcs and azblob syncs; requests will be made at this interval. Defaults to 5 seconds.                                                                                                              |
 | tls         | optional `boolean` | Enable/Disable secure TLS connectivity. Currently used only by gRPC sync. Default (ex: if unset) is false, which will use an insecure connection                                                                 |
 | providerID  | optional `string`  | Value binds to grpc connection's providerID field. gRPC server implementations may use this to identify connecting flagd instance                                                                                |
 | selector    | optional `string`  | Value binds to grpc connection's selector field. gRPC server implementations may use this to filter flag configurations                                                                                          |
@@ -65,6 +66,7 @@ Sync providers:
 - `grpc`(insecure) - grpc-source:8080
 - `grpcs`(secure) - my-flag-source:8080
 - `gcs` - gs://my-bucket/my-flags.json
+- `azblob` - azblob://my-container/my-flags.json
 
 Startup command:
 
@@ -80,7 +82,8 @@ Startup command:
             {"uri":"grpc-source:8080","provider":"grpc"},
             {"uri":"my-flag-source:8080","provider":"grpc", "maxMsgSize": 5242880},
             {"uri":"my-flag-source:8080","provider":"grpc", "certPath": "/certs/ca.cert", "tls": true, "providerID": "flagd-weatherapp-sidecar", "selector": "source=database,app=weatherapp"},
-            {"uri":"gs://my-bucket/my-flag.json","provider":"gcs"}]'
+            {"uri":"gs://my-bucket/my-flag.json","provider":"gcs"},
+            {"uri":"azblob://my-container/my-flag.json","provider":"azblob"}]'
 ```
 
 Configuration file,
@@ -111,4 +114,6 @@ sources:
     selector: "source=database,app=weatherapp"
   - uri: gs://my-bucket/my-flag.json
     provider: gcs
+  - uri: azblob://my-container/my-flags.json
+    provider: azblob
 ```