Add storage based ingester to m3coordinator

glide.yaml

@@ -31,6 +31,9 @@ import:
   - package: github.com/m3db/m3ctl
     version: acc762bfdd42ecb192d34e48fa7ca1fd7ee088ac
 
+  - package: github.com/m3db/m3msg
+    version: 4851e2719e06b15f1fc247e1d00339192963990e
+
   - package: github.com/m3db/bitset
     version: 07973db6b78acb62ac207d0538055e874b49d90d
 

scripts/development/m3_stack/docker-compose.yml

@@ -37,9 +37,11 @@ services:
     expose:
       - "7201"
       - "7203"
+      - "7507"
     ports:
       - "0.0.0.0:7201:7201"
       - "0.0.0.0:7203:7203"
+      - "0.0.0.0:7507:7507"
     networks:
       - backend
     build:

scripts/development/m3_stack/m3coordinator.yml

@@ -45,3 +45,18 @@ clusters:
         jitter: true
       backgroundHealthCheckFailLimit: 4
       backgroundHealthCheckFailThrottleFactor: 0.5
+
+ingest:
+  ingester:
+    workerPoolSize: 100
+    opPool:
+      size: 100
+    retry:
+      maxRetries: 3
+      jitter: true
+  m3msg:
+    server:
+      listenAddress: 0.0.0.0:7507
+      retry:
+        maxBackoff: 10s
+        jitter: true

src/cmd/services/m3coordinator/downsample/flush_handler.go

@@ -34,6 +34,7 @@ import (
 	"github.com/m3db/m3metrics/metric/aggregated"
 	"github.com/m3db/m3x/instrument"
 	xsync "github.com/m3db/m3x/sync"
+	xtime "github.com/m3db/m3x/time"
 
 	"github.com/uber-go/tally"
 )
@@ -142,7 +143,7 @@ func (w *downsamplerFlushHandlerWriter) Write(
 				Timestamp: time.Unix(0, mp.TimeNanos),
 				Value:     mp.Value,
 			}},
-			Unit: mp.StoragePolicy.Resolution().Precision,
+			Unit: xtime.Millisecond,
 			Attributes: storage.Attributes{
 				MetricsType: storage.AggregatedMetricsType,
 				Retention:   mp.StoragePolicy.Retention().Duration(),

src/cmd/services/m3coordinator/ingest/config.go

@@ -0,0 +1,85 @@
+// Copyright (c) 2018 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package ingest
+
+import (
+	"github.com/m3db/m3/src/query/storage"
+	"github.com/m3db/m3/src/x/serialize"
+	"github.com/m3db/m3x/instrument"
+	"github.com/m3db/m3x/pool"
+	"github.com/m3db/m3x/retry"
+	xsync "github.com/m3db/m3x/sync"
+)
+
+// Configuration configs the ingester.
+type Configuration struct {
+	WorkerPoolSize int                          `yaml:"workerPoolSize"`
+	OpPool         pool.ObjectPoolConfiguration `yaml:"opPool"`
+	Retry          retry.Configuration          `yaml:"retry"`
+}
+
+// NewIngester creates an ingester with an appender.
+func (cfg Configuration) NewIngester(
+	appender storage.Appender,
+	instrumentOptions instrument.Options,
+) (*Ingester, error) {
+	opts, err := cfg.newOptions(appender, instrumentOptions)
+	if err != nil {
+		return nil, err
+	}
+	return NewIngester(*opts), nil
+}
+
+func (cfg Configuration) newOptions(
+	appender storage.Appender,
+	instrumentOptions instrument.Options,
+) (*Options, error) {
+	scope := instrumentOptions.MetricsScope().Tagged(
+		map[string]string{"component": "ingester"},
+	)
+	workers, err := xsync.NewPooledWorkerPool(
+		cfg.WorkerPoolSize,
+		xsync.NewPooledWorkerPoolOptions().
+			SetInstrumentOptions(instrumentOptions),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	workers.Init()
+	tagDecoderPool := serialize.NewTagDecoderPool(
+		serialize.NewTagDecoderOptions(),
+		pool.NewObjectPoolOptions().
+			SetInstrumentOptions(instrumentOptions.
+				SetMetricsScope(instrumentOptions.MetricsScope().
+					SubScope("tag-decoder-pool"))),
+	)
+	tagDecoderPool.Init()
+	opts := Options{
+		Appender:          appender,
+		Workers:           workers,
+		PoolOptions:       cfg.OpPool.NewObjectPoolOptions(instrumentOptions),
+		TagDecoderPool:    tagDecoderPool,
+		RetryOptions:      cfg.Retry.NewOptions(scope),
+		InstrumentOptions: instrumentOptions,
+	}
+	return &opts, nil
+}

src/cmd/services/m3coordinator/ingest/ingest.go

@@ -0,0 +1,196 @@
+// Copyright (c) 2018 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package ingest
+
+import (
+	"context"
+	"time"
+
+	"github.com/m3db/m3/src/cmd/services/m3coordinator/server/m3msg"
+	"github.com/m3db/m3/src/query/models"
+	"github.com/m3db/m3/src/query/storage"
+	"github.com/m3db/m3/src/query/ts"
+	"github.com/m3db/m3/src/x/common"
+	"github.com/m3db/m3/src/x/serialize"
+	"github.com/m3db/m3metrics/metric/id"
+	"github.com/m3db/m3metrics/policy"
+	xerrors "github.com/m3db/m3x/errors"
+	"github.com/m3db/m3x/instrument"
+	"github.com/m3db/m3x/pool"
+	"github.com/m3db/m3x/retry"
+	xsync "github.com/m3db/m3x/sync"
+
+	"github.com/uber-go/tally"
+)
+
+// Options configures the ingester.
+type Options struct {
+	Appender          storage.Appender
+	Workers           xsync.PooledWorkerPool
+	PoolOptions       pool.ObjectPoolOptions
+	TagDecoderPool    serialize.TagDecoderPool
+	RetryOptions      retry.Options
+	InstrumentOptions instrument.Options
+}
+
+type ingestMetrics struct {
+	ingestError   tally.Counter
+	ingestSuccess tally.Counter
+}
+
+func newIngestMetrics(scope tally.Scope) ingestMetrics {
+	return ingestMetrics{
+		ingestError:   scope.Counter("ingest-error"),
+		ingestSuccess: scope.Counter("ingest-success"),
+	}
+}
+
+// Ingester ingests metrics with a worker pool.
+type Ingester struct {
+	workers xsync.PooledWorkerPool
+	p       pool.ObjectPool
+}
+
+// NewIngester creates an ingester.
+func NewIngester(
+	opts Options,
+) *Ingester {
+	retrier := retry.NewRetrier(opts.RetryOptions)
+	m := newIngestMetrics(opts.InstrumentOptions.MetricsScope())
+	p := pool.NewObjectPool(opts.PoolOptions)
+	p.Init(
+		func() interface{} {
+			// NB: we don't need a pool for the tag decoder since the ops are
+			// pooled, but currently this is the only way to get tag decoder.
+			tagDecoder := opts.TagDecoderPool.Get()
+			op := ingestOp{
+				s:  opts.Appender,
+				r:  retrier,
+				it: serialize.NewMetricTagsIterator(tagDecoder, nil),
+				p:  p,
+				m:  m,
+			}
+			op.attemptFn = op.attempt
+			op.ingestFn = op.ingest
+			return &op
+		},
+	)
+	return &Ingester{
+		workers: opts.Workers,
+		p:       p,
+	}
+}
+
+// Ingest ingests a metric asynchronously with callback.
+func (i *Ingester) Ingest(
+	ctx context.Context,
+	id []byte,
+	metricTime time.Time,
+	value float64,
+	sp policy.StoragePolicy,
+	callback *m3msg.RefCountedCallback,
+) {
+	op := i.p.Get().(*ingestOp)
+	op.c = ctx
+	op.id = id
+	op.metricTime = metricTime
+	op.value = value
+	op.sp = sp
+	op.callback = callback
+	i.workers.Go(op.ingestFn)
+}
+
+type ingestOp struct {
+	s         storage.Appender
+	r         retry.Retrier
+	it        id.SortedTagIterator
+	p         pool.ObjectPool
+	m         ingestMetrics
+	attemptFn retry.Fn
+	ingestFn  func()
+
+	c          context.Context
+	id         []byte
+	metricTime time.Time
+	value      float64
+	sp         policy.StoragePolicy
+	callback   *m3msg.RefCountedCallback
+	q          storage.WriteQuery
+}
+
+func (op *ingestOp) ingest() {
+	if err := op.resetWriteQuery(); err != nil {
+		op.m.ingestError.Inc(1)
+		op.callback.Callback(m3msg.OnRetriableError)
+		op.p.Put(op)
+		return
+	}
+	if err := op.r.Attempt(op.attemptFn); err != nil {
+		if xerrors.IsNonRetryableError(err) {
+			op.callback.Callback(m3msg.OnNonRetriableError)
+		} else {
+			op.callback.Callback(m3msg.OnRetriableError)
+		}
+		op.m.ingestError.Inc(1)
+		op.p.Put(op)
+		return
+	}
+	op.m.ingestSuccess.Inc(1)
+	op.callback.Callback(m3msg.OnSuccess)
+	op.p.Put(op)
+}
+
+func (op *ingestOp) attempt() error {
+	return op.s.Write(op.c, &op.q)
+}
+
+func (op *ingestOp) resetWriteQuery() error {
+	if err := op.resetTags(); err != nil {
+		return err
+	}
+	op.resetDataPoints()
+	op.q.Unit = common.SanitizeUnitForM3DB(op.sp.Resolution().Precision)
+	op.q.Attributes.MetricsType = storage.AggregatedMetricsType
+	op.q.Attributes.Resolution = op.sp.Resolution().Window
+	op.q.Attributes.Retention = op.sp.Retention().Duration()
+	return nil
+}
+
+func (op *ingestOp) resetTags() error {
+	op.it.Reset(op.id)
+	op.q.Tags = op.q.Tags[:0]
+	for op.it.Next() {
+		name, value := op.it.Current()
+		op.q.Tags = append(op.q.Tags, models.Tag{
+			Name:  append([]byte(nil), name...),
+			Value: append([]byte(nil), value...),
+		})
+	}
+	return op.it.Err()
+}
+
+func (op *ingestOp) resetDataPoints() {
+	if len(op.q.Datapoints) != 1 {
+		op.q.Datapoints = make(ts.Datapoints, 1)
+	}
+	op.q.Datapoints[0].Timestamp = op.metricTime
+	op.q.Datapoints[0].Value = op.value
+}