add flag to allow broker (and for on-prem the consumers) to avoid actual store health checks by returning healthy for the checks

Author: ddowker

.github/workflows/ci-workflow.yaml

@@ -38,7 +38,7 @@ jobs:
       - uses: actions/setup-go@v4
         with:
           go-version-file: 'go.mod'
-      - run: sudo apt install -y protobuf-compiler librocksdb-dev libsqlite3-dev etcd-server
+      - run: sudo apt-get update && sudo apt-get install -y protobuf-compiler librocksdb-dev libsqlite3-dev etcd-server 
       - run: |
           sha=${{ github.sha }}
           if [[ '${{ github.ref }}' == 'refs/heads/arize' ]]; then

broker/stores/health_check.go

@@ -90,6 +90,10 @@ func runCheck(ctx context.Context, s *ActiveStore) error {
 		testContent = "health-check\n"
 	)
 
+	if ForceStoreHealthCheckToHealthy {
+		return nil
+	}
+
 	// 1. PUT test file
 	var content = strings.NewReader(testContent)
 	if err := s.Store.Put(ctx, testPath, content, int64(len(testContent)), ""); err != nil {

broker/stores/stores.go

@@ -17,6 +17,9 @@ var (
 
 	// Whether to return an unsigned URL when a signed URL is requested. Useful when clients do not require the signing.
 	DisableSignedUrls bool = false
+
+	// Whether to force the health check of fragment stores to healthy.
+	ForceStoreHealthCheckToHealthy bool = false
 )
 
 // RegisterProviders registers store constructors for different storage schemes.

cmd/gazette/main.go

@@ -39,16 +39,17 @@ const iniFilename = "gazette.ini"
 var Config = new(struct {
 	Broker struct {
 		mbp.ServiceConfig
-		Limit             uint32        `long:"limit" env:"LIMIT" default:"1024" description:"Maximum number of Journals the broker will allocate"`
-		FileRoot          string        `long:"file-root" env:"FILE_ROOT" description:"Local path which roots file:// fragment stores (optional)"`
-		FileOnly          bool          `long:"file-only" env:"FILE_ONLY" description:"Use the local file:// store for all journal fragments, ignoring cloud bucket storage configuration (for example, S3)"`
-		MaxAppendRate     uint32        `long:"max-append-rate" env:"MAX_APPEND_RATE" default:"0" description:"Max rate (in bytes-per-sec) that any one journal may be appended to. If zero, there is no max rate"`
-		MaxReplication    uint32        `long:"max-replication" env:"MAX_REPLICATION" default:"9" description:"Maximum effective replication of any one journal, which upper-bounds its stated replication."`
-		MinAppendRate     uint32        `long:"min-append-rate" env:"MIN_APPEND_RATE" default:"65536" description:"Min rate (in bytes-per-sec) at which a client may stream Append RPC content. RPCs unable to sustain this rate are aborted"`
-		WatchDelay        time.Duration `long:"watch-delay" env:"WATCH_DELAY" default:"30ms" description:"Delay applied to the application of watched Etcd events. Larger values amortize the processing of fast-changing Etcd keys."`
-		AuthKeys          string        `long:"auth-keys" env:"AUTH_KEYS" description:"Whitespace or comma separated, base64-encoded keys used to sign (first key) and verify (all keys) Authorization tokens." json:"-"`
-		AutoSuspend       bool          `long:"auto-suspend" env:"AUTO_SUSPEND" description:"Automatically suspend journals which have persisted all fragments"`
-		DisableSignedUrls bool          `long:"disable-signed-urls" env:"DISABLE_SIGNED_URLS" description:"When a signed URL is requested, return an unsigned URL instead. This is useful when clients do not require the signing."`
+		Limit                          uint32        `long:"limit" env:"LIMIT" default:"1024" description:"Maximum number of Journals the broker will allocate"`
+		FileRoot                       string        `long:"file-root" env:"FILE_ROOT" description:"Local path which roots file:// fragment stores (optional)"`
+		FileOnly                       bool          `long:"file-only" env:"FILE_ONLY" description:"Use the local file:// store for all journal fragments, ignoring cloud bucket storage configuration (for example, S3)"`
+		MaxAppendRate                  uint32        `long:"max-append-rate" env:"MAX_APPEND_RATE" default:"0" description:"Max rate (in bytes-per-sec) that any one journal may be appended to. If zero, there is no max rate"`
+		MaxReplication                 uint32        `long:"max-replication" env:"MAX_REPLICATION" default:"9" description:"Maximum effective replication of any one journal, which upper-bounds its stated replication."`
+		MinAppendRate                  uint32        `long:"min-append-rate" env:"MIN_APPEND_RATE" default:"65536" description:"Min rate (in bytes-per-sec) at which a client may stream Append RPC content. RPCs unable to sustain this rate are aborted"`
+		WatchDelay                     time.Duration `long:"watch-delay" env:"WATCH_DELAY" default:"30ms" description:"Delay applied to the application of watched Etcd events. Larger values amortize the processing of fast-changing Etcd keys."`
+		AuthKeys                       string        `long:"auth-keys" env:"AUTH_KEYS" description:"Whitespace or comma separated, base64-encoded keys used to sign (first key) and verify (all keys) Authorization tokens." json:"-"`
+		AutoSuspend                    bool          `long:"auto-suspend" env:"AUTO_SUSPEND" description:"Automatically suspend journals which have persisted all fragments"`
+		DisableSignedUrls              bool          `long:"disable-signed-urls" env:"DISABLE_SIGNED_URLS" description:"When a signed URL is requested, return an unsigned URL instead. This is useful when clients do not require the signing."`
+		ForceStoreHealthCheckToHealthy bool          `long:"force-store-health-check-to-healthy" env:"FORCE_STORE_HEALTH_CHECK_TO_HEALTHY" description:"Force the health check of fragment stores to healthy"`
 	} `group:"Broker" namespace:"broker" env-namespace:"BROKER"`
 
 	Etcd struct {
@@ -151,6 +152,7 @@ func (cmdServe) Execute(args []string) error {
 	broker.MinAppendRate = int64(Config.Broker.MinAppendRate)
 	pb.MaxReplication = int32(Config.Broker.MaxReplication)
 	stores.DisableSignedUrls = Config.Broker.DisableSignedUrls
+	stores.ForceStoreHealthCheckToHealthy = Config.Broker.ForceStoreHealthCheckToHealthy
 
 	var (
 		lo   = pb.NewAuthJournalClient(pb.NewJournalClient(srv.GRPCLoopback), authorizer)

mainboilerplate/runconsumer/run_consumer.go

@@ -92,13 +92,14 @@ type Config interface {
 type BaseConfig struct {
 	Consumer struct {
 		mbp.ServiceConfig
-		Limit                  uint32        `long:"limit" env:"LIMIT" default:"32" description:"Maximum number of Shards this consumer process will allocate"`
-		MaxHotStandbys         uint32        `long:"max-hot-standbys" env:"MAX_HOT_STANDBYS" default:"3" description:"Maximum effective hot standbys of any one shard, which upper-bounds its stated hot-standbys."`
-		WatchDelay             time.Duration `long:"watch-delay" env:"WATCH_DELAY" default:"30ms" description:"Delay applied to the application of watched Etcd events. Larger values amortize the processing of fast-changing Etcd keys."`
-		SkipSignedURLs         bool          `long:"skip-signed-urls" env:"SKIP_SIGNED_URLS" description:"When a signed URL is received, use fragment info instead to retrieve data with auth header. This is useful when clients do not wish/require the signing."`
-		AuthKeys               string        `long:"auth-keys" env:"AUTH_KEYS" description:"Whitespace or comma separated, base64-encoded keys used to sign (first key) and verify (all keys) Authorization tokens." json:"-"`
-		AWSAccessKeyIDPath     string        `long:"aws-access-key-id-path" env:"AWS_ACCESS_KEY_ID_PATH" default:"" description:"file path to the aws access key id secret"`
-		AWSSecretAccessKeyPath string        `long:"aws-secret-access-key-path" env:"AWS_SECRET_ACCESS_KEY_PATH" default:"" description:"file path to the aws secret access key secret"`
+		Limit                          uint32        `long:"limit" env:"LIMIT" default:"32" description:"Maximum number of Shards this consumer process will allocate"`
+		MaxHotStandbys                 uint32        `long:"max-hot-standbys" env:"MAX_HOT_STANDBYS" default:"3" description:"Maximum effective hot standbys of any one shard, which upper-bounds its stated hot-standbys."`
+		WatchDelay                     time.Duration `long:"watch-delay" env:"WATCH_DELAY" default:"30ms" description:"Delay applied to the application of watched Etcd events. Larger values amortize the processing of fast-changing Etcd keys."`
+		SkipSignedURLs                 bool          `long:"skip-signed-urls" env:"SKIP_SIGNED_URLS" description:"When a signed URL is received, use fragment info instead to retrieve data with auth header. This is useful when clients do not wish/require the signing."`
+		AuthKeys                       string        `long:"auth-keys" env:"AUTH_KEYS" description:"Whitespace or comma separated, base64-encoded keys used to sign (first key) and verify (all keys) Authorization tokens." json:"-"`
+		AWSAccessKeyIDPath             string        `long:"aws-access-key-id-path" env:"AWS_ACCESS_KEY_ID_PATH" default:"" description:"file path to the aws access key id secret"`
+		AWSSecretAccessKeyPath         string        `long:"aws-secret-access-key-path" env:"AWS_SECRET_ACCESS_KEY_PATH" default:"" description:"file path to the aws secret access key secret"`
+		ForceStoreHealthCheckToHealthy bool          `long:"force-store-health-check-to-healthy" env:"FORCE_STORE_HEALTH_CHECK_TO_HEALTHY" description:"Force the health check of fragment stores to healthy"`
 	} `group:"Consumer" namespace:"consumer" env-namespace:"CONSUMER"`
 
 	Broker struct {
@@ -196,6 +197,10 @@ func (sc Cmd) Execute(args []string) error {
 		stores.DisableSignedUrls = true
 	}
 
+	if bc.Consumer.ForceStoreHealthCheckToHealthy {
+		stores.ForceStoreHealthCheckToHealthy = true
+	}
+
 	// Load AWS credentials from file paths if both are configured.
 	if bc.Consumer.AWSAccessKeyIDPath != "" && bc.Consumer.AWSSecretAccessKeyPath != "" {
 		accessKeyID, err := os.ReadFile(bc.Consumer.AWSAccessKeyIDPath)