[dbnode] Add namespace runtime options for runtime per-namespace config changes

src/dbnode/generated/proto/namespace/namespace.proto

@@ -1,6 +1,8 @@
 syntax = "proto3";
+
 package namespace;
 
+import "google/protobuf/wrappers.proto";
 import "github.com/m3db/m3/src/dbnode/generated/proto/namespace/schema.proto";
 
 message RetentionOptions {
@@ -19,18 +21,24 @@ message IndexOptions {
 }
 
 message NamespaceOptions {
-    bool bootstrapEnabled             = 1;
-    bool flushEnabled                 = 2;
-    bool writesToCommitLog            = 3;
-    bool cleanupEnabled               = 4;
-    bool repairEnabled                = 5;
-    RetentionOptions retentionOptions = 6;
-    bool snapshotEnabled              = 7;
-    IndexOptions indexOptions         = 8;
-    SchemaOptions schemaOptions       = 9;
-    bool coldWritesEnabled            = 10;
+    bool bootstrapEnabled                  = 1;
+    bool flushEnabled                      = 2;
+    bool writesToCommitLog                 = 3;
+    bool cleanupEnabled                    = 4;
+    bool repairEnabled                     = 5;
+    RetentionOptions retentionOptions      = 6;
+    bool snapshotEnabled                   = 7;
+    IndexOptions indexOptions              = 8;
+    SchemaOptions schemaOptions            = 9;
+    bool coldWritesEnabled                 = 10;
+    NamespaceRuntimeOptions runtimeOptions = 11;
 }
 
 message Registry {
     map<string, NamespaceOptions> namespaces = 1;
 }
+
+message NamespaceRuntimeOptions {
+    google.protobuf.DoubleValue writeIndexingPerCPUConcurrency = 1;
+    google.protobuf.DoubleValue flushIndexingPerCPUConcurrency = 2;
+}

src/dbnode/namespace/convert.go

@@ -28,6 +28,8 @@ import (
 	"github.com/m3db/m3/src/dbnode/retention"
 	"github.com/m3db/m3/src/x/ident"
 	xtime "github.com/m3db/m3/src/x/time"
+
+	protobuftypes "github.com/gogo/protobuf/types"
 )
 
 var (
@@ -80,6 +82,25 @@ func ToIndexOptions(
 	return iopts, nil
 }
 
+// ToRuntimeOptions converts nsproto.NamespaceRuntimeOptions to RuntimeOptions.
+func ToRuntimeOptions(
+	opts *nsproto.NamespaceRuntimeOptions,
+) (RuntimeOptions, error) {
+	runtimeOpts := NewRuntimeOptions()
+	if opts == nil {
+		return runtimeOpts, nil
+	}
+	if v := opts.WriteIndexingPerCPUConcurrency; v != nil {
+		newValue := v.Value
+		runtimeOpts = runtimeOpts.SetWriteIndexingPerCPUConcurrency(&newValue)
+	}
+	if v := opts.FlushIndexingPerCPUConcurrency; v != nil {
+		newValue := v.Value
+		runtimeOpts = runtimeOpts.SetFlushIndexingPerCPUConcurrency(&newValue)
+	}
+	return runtimeOpts, nil
+}
+
 // ToMetadata converts nsproto.Options to Metadata
 func ToMetadata(
 	id string,
@@ -104,6 +125,11 @@ func ToMetadata(
 		return nil, err
 	}
 
+	runtimeOpts, err := ToRuntimeOptions(opts.RuntimeOptions)
+	if err != nil {
+		return nil, err
+	}
+
 	mopts := NewOptions().
 		SetBootstrapEnabled(opts.BootstrapEnabled).
 		SetFlushEnabled(opts.FlushEnabled).
@@ -114,7 +140,12 @@ func ToMetadata(
 		SetSchemaHistory(sr).
 		SetRetentionOptions(ropts).
 		SetIndexOptions(iopts).
-		SetColdWritesEnabled(opts.ColdWritesEnabled)
+		SetColdWritesEnabled(opts.ColdWritesEnabled).
+		SetRuntimeOptions(runtimeOpts)
+
+	if err := mopts.Validate(); err != nil {
+		return nil, err
+	}
 
 	return NewMetadata(ident.StringID(id), mopts)
 }
@@ -172,5 +203,31 @@ func OptionsToProto(opts Options) *nsproto.NamespaceOptions {
 			BlockSizeNanos: iopts.BlockSize().Nanoseconds(),
 		},
 		ColdWritesEnabled: opts.ColdWritesEnabled(),
+		RuntimeOptions:    toRuntimeOptions(opts.RuntimeOptions()),
+	}
+}
+
+// toRuntimeOptions returns the corresponding RuntimeOptions proto.
+func toRuntimeOptions(opts RuntimeOptions) *nsproto.NamespaceRuntimeOptions {
+	if opts == nil || opts.IsDefaults() {
+		return nil
+	}
+	var (
+		writeIndexingPerCPUConcurrency *protobuftypes.DoubleValue
+		flushIndexingPerCPUConcurrency *protobuftypes.DoubleValue
+	)
+	if v := opts.WriteIndexingPerCPUConcurrency(); v != nil {
+		writeIndexingPerCPUConcurrency = &protobuftypes.DoubleValue{
+			Value: *v,
+		}
+	}
+	if v := opts.FlushIndexingPerCPUConcurrency(); v != nil {
+		flushIndexingPerCPUConcurrency = &protobuftypes.DoubleValue{
+			Value: *v,
+		}
+	}
+	return &nsproto.NamespaceRuntimeOptions{
+		WriteIndexingPerCPUConcurrency: writeIndexingPerCPUConcurrency,
+		FlushIndexingPerCPUConcurrency: flushIndexingPerCPUConcurrency,
 	}
 }

src/dbnode/namespace/dynamic.go

@@ -199,11 +199,13 @@ func (r *dynamicRegistry) run() {
 
 		if m.Equal(r.maps()) {
 			r.metrics.numInvalidUpdates.Inc(1)
-			r.logger.Warn("dynamic namespace registry received identical update, skipping")
+			r.logger.Warn("dynamic namespace registry received identical update, skipping",
+				zap.Int("version", val.Version()))
 			continue
 		}
 
-		r.logger.Info("dynamic namespace registry updated to version", zap.Int("version", val.Version()))
+		r.logger.Info("dynamic namespace registry updated to version",
+			zap.Int("version", val.Version()))
 		r.Lock()
 		r.currentValue = val
 		r.currentMap = m