Allow AggregationOptions to be set for a namespace.

pkg/apis/m3dboperator/v1alpha1/namespace.go

@@ -63,6 +63,41 @@ type IndexOptions struct {
 	BlockSize string `json:"blockSize,omitempty"`
 }
 
+// AggregationOptions is a set of options for aggregating data
+// within the namespace.
+type AggregationOptions struct {
+	// Aggregations are the aggregations for a namespace.
+	Aggregations []Aggregation `json:"aggregations,omitempty"`
+}
+
+// Aggregation describes data points within a namespace.
+type Aggregation struct {
+	// Aggregated indicates whether data points are aggregated or not.
+	Aggregated bool `json:"aggregated,omitempty"`
+
+	// Attributes defines how data is aggregated when Aggregated is set to true.
+	// This field is ignored when aggregated is false.
+	Attributes AggregatedAttributes `json:"attributes,omitempty"`
+}
+
+// AggregatedAttributes are attributes specifying how data points are aggregated.
+type AggregatedAttributes struct {
+	// Resolution is the time range to aggregate data across.
+	Resolution string `json:"resolution,omitempty"`
+
+	// DownsampleOptions stores options for downsampling data points.
+	DownsampleOptions *DownsampleOptions `json:"downsampleOptions,omitempty"`
+}
+
+// DownsampleOptions is a set of options related to downsampling data.
+type DownsampleOptions struct {
+	// All indicates whether to send data points to this namespace.
+	// If set to false, this namespace will not receive data points. In this
+	// case, data will need to be sent to the namespace via another mechanism
+	// (e.g. rollup/recording rules).
+	All bool `json:"all,omitempty"`
+}
+
 // NamespaceOptions defines parameters for an M3DB namespace. See
 // https://m3db.github.io/m3/operational_guide/namespace_configuration/ for more
 // details.
@@ -93,4 +128,7 @@ type NamespaceOptions struct {
 
 	// ColdWritesEnabled controls whether cold writes are enabled.
 	ColdWritesEnabled bool `json:"coldWritesEnabled,omitempty"`
+
+	// AggregationOptions sets the aggregation parameters.
+	AggregationOptions AggregationOptions `json:"aggregationOptions,omitempty"`
 }

pkg/m3admin/namespace/namespace.go

@@ -90,16 +90,22 @@ func m3dbNamespaceOptsFromSpec(opts *myspec.NamespaceOptions) (*m3ns.NamespaceOp
 		return nil, err
 	}
 
+	aggOpts, err := m3dbAggregationOptsFromSpec(opts.AggregationOptions)
+	if err != nil {
+		return nil, err
+	}
+
 	return &m3ns.NamespaceOptions{
-		BootstrapEnabled:  opts.BootstrapEnabled,
-		FlushEnabled:      opts.FlushEnabled,
-		WritesToCommitLog: opts.WritesToCommitLog,
-		CleanupEnabled:    opts.CleanupEnabled,
-		RepairEnabled:     opts.RepairEnabled,
-		RetentionOptions:  retentionOpts,
-		SnapshotEnabled:   opts.SnapshotEnabled,
-		IndexOptions:      indexOpts,
-		ColdWritesEnabled: opts.ColdWritesEnabled,
+		BootstrapEnabled:   opts.BootstrapEnabled,
+		FlushEnabled:       opts.FlushEnabled,
+		WritesToCommitLog:  opts.WritesToCommitLog,
+		CleanupEnabled:     opts.CleanupEnabled,
+		RepairEnabled:      opts.RepairEnabled,
+		RetentionOptions:   retentionOpts,
+		SnapshotEnabled:    opts.SnapshotEnabled,
+		IndexOptions:       indexOpts,
+		ColdWritesEnabled:  opts.ColdWritesEnabled,
+		AggregationOptions: aggOpts,
 	}, nil
 }
 
@@ -150,3 +156,39 @@ func m3dbIndexOptsFromSpec(opts myspec.IndexOptions) (*m3ns.IndexOptions, error)
 		BlockSizeNanos: blockSize.Nanoseconds(),
 	}, nil
 }
+
+func m3dbAggregationOptsFromSpec(opts myspec.AggregationOptions) (*m3ns.AggregationOptions, error) {
+	if len(opts.Aggregations) == 0 {
+		return nil, nil
+	}
+
+	aggs := make([]*m3ns.Aggregation, 0, len(opts.Aggregations))
+	for _, specAgg := range opts.Aggregations {
+		agg := &m3ns.Aggregation{Aggregated: specAgg.Aggregated}
+		if agg.Aggregated {
+			resolution, err := time.ParseDuration(specAgg.Attributes.Resolution)
+			if err != nil {
+				return nil, fmt.Errorf("failed to parse aggregation option Resolution: %w", err)
+			}
+
+			agg.Attributes = &m3ns.AggregatedAttributes{
+				ResolutionNanos: resolution.Nanoseconds(),
+			}
+
+			if specAgg.Attributes.DownsampleOptions == nil {
+				agg.Attributes.DownsampleOptions = &m3ns.DownsampleOptions{All: true}
+			} else {
+				agg.Attributes.DownsampleOptions = &m3ns.DownsampleOptions{
+					All: specAgg.Attributes.DownsampleOptions.All,
+				}
+			}
+		}
+
+		aggs = append(aggs, agg)
+	}
+
+	return &m3ns.AggregationOptions{
+		Aggregations: aggs,
+	}, nil
+
+}

pkg/m3admin/namespace/namespace_test.go

@@ -79,6 +79,11 @@ func TestRequestFromSpec(t *testing.T) {
 						BlockSize: "1s",
 						Enabled:   true,
 					},
+					AggregationOptions: myspec.AggregationOptions{
+						Aggregations: []myspec.Aggregation{
+							{Aggregated: false},
+						},
+					},
 				},
 			},
 			req: &admin.NamespaceAddRequest{
@@ -97,6 +102,70 @@ func TestRequestFromSpec(t *testing.T) {
 						BlockSizeNanos: 1000000000,
 						Enabled:        true,
 					},
+					AggregationOptions: &m3ns.AggregationOptions{
+						Aggregations: []*m3ns.Aggregation{
+							{Aggregated: false},
+						},
+					},
+				},
+			},
+		},
+		{
+			ns: myspec.Namespace{
+				Name: "aggregated",
+				Options: &myspec.NamespaceOptions{
+					BootstrapEnabled: true,
+					RetentionOptions: myspec.RetentionOptions{
+						RetentionPeriod:                     "1s",
+						BlockSize:                           "1s",
+						BufferFuture:                        "1s",
+						BufferPast:                          "1s",
+						BlockDataExpiry:                     true,
+						BlockDataExpiryAfterNotAccessPeriod: "1s",
+					},
+					IndexOptions: myspec.IndexOptions{
+						BlockSize: "1s",
+						Enabled:   true,
+					},
+					AggregationOptions: myspec.AggregationOptions{
+						Aggregations: []myspec.Aggregation{
+							{
+								Aggregated: true,
+								Attributes: myspec.AggregatedAttributes{
+									Resolution: "1s",
+								},
+							},
+						},
+					},
+				},
+			},
+			req: &admin.NamespaceAddRequest{
+				Name: "aggregated",
+				Options: &m3ns.NamespaceOptions{
+					BootstrapEnabled: true,
+					RetentionOptions: &m3ns.RetentionOptions{
+						RetentionPeriodNanos:                     1000000000,
+						BlockSizeNanos:                           1000000000,
+						BufferFutureNanos:                        1000000000,
+						BufferPastNanos:                          1000000000,
+						BlockDataExpiry:                          true,
+						BlockDataExpiryAfterNotAccessPeriodNanos: 1000000000,
+					},
+					IndexOptions: &m3ns.IndexOptions{
+						BlockSizeNanos: 1000000000,
+						Enabled:        true,
+					},
+					AggregationOptions: &m3ns.AggregationOptions{
+						Aggregations: []*m3ns.Aggregation{
+							{
+								Aggregated: true,
+								Attributes: &m3ns.AggregatedAttributes{
+									ResolutionNanos:   1000000000,
+									DownsampleOptions: &m3ns.DownsampleOptions{All: true},
+								},
+							},
+						},
+					},
 				},
 			},
 		},

pkg/m3admin/namespace/presets.go

@@ -62,6 +62,16 @@ var (
 			BlockSize: (2 * time.Hour).String(),
 		},
 		ColdWritesEnabled: false,
+		AggregationOptions: myspec.AggregationOptions{
+			Aggregations: []myspec.Aggregation{
+				{
+					Aggregated: true,
+					Attributes: myspec.AggregatedAttributes{
+						Resolution: (10 * time.Second).String(),
+					},
+				},
+			},
+		},
 	}
 
 	presetOneMinuteFourtyDaysIndexed = myspec.NamespaceOptions{
@@ -84,5 +94,15 @@ var (
 			BlockSize: (24 * time.Hour).String(),
 		},
 		ColdWritesEnabled: false,
+		AggregationOptions: myspec.AggregationOptions{
+			Aggregations: []myspec.Aggregation{
+				{
+					Aggregated: true,
+					Attributes: myspec.AggregatedAttributes{
+						Resolution: (1 * time.Minute).String(),
+					},
+				},
+			},
+		},
 	}
 )