[aggregator] m3aggregator with pass-through functionality #2105

fishie9 · 2020-01-13T21:39:22Z

What this PR does / why we need it:

This PR introduces the pass-through functionality into m3aggregator.

Special notes for your reviewer:
This is a PR squashing commits in #1802

Does this PR introduce a user-facing and/or backwards incompatible change?:

Does this PR require updating code package or user-facing documentation?:

prateek · 2020-01-13T21:47:51Z

glide.lock

@@ -278,7 +278,7 @@ imports:
  subpackages:
  - regexp
 - name: github.com/magiconair/properties
-  version: de8848e004dd33dc07a2947b3d76f618a7fc7ef1
+  version: 7757cc9fdb852f7579b24170bcacda2c7471bb6a


do you need this change?

src/aggregator/aggregator/aggregator.go

+	callStart := agg.nowFn()
+	agg.metrics.passThrough.Inc(1)
+
+	if agg.state != aggregatorOpen {


src/aggregator/aggregator/aggregator.go

prateek · 2020-01-13T21:55:06Z

src/aggregator/aggregator/aggregator.go

@@ -74,6 +76,9 @@ type Aggregator interface {
 	// AddForwarded adds a forwarded metric with metadata.
 	AddForwarded(metric aggregated.ForwardedMetric, metadata metadata.ForwardMetadata) error

+	// AddPassThrough add a pass-through metric with metadata


nit: period at the end of the comment.

prateek · 2020-01-13T21:56:28Z

src/aggregator/integration/client.go

+	metric aggregated.Metric,
+	metadata metadata.TimedMetadata,
+) error {
+	// Siyu?


do you need to delete this?

src/cmd/services/m3aggregator/main/main.go

src/cmd/services/m3aggregator/serve/serve.go

src/cmd/services/m3aggregator/config/aggregator.go

prateek · 2020-01-13T22:17:46Z

src/cmd/services/m3aggregator/config/aggregator.go

+// passThroughWriter writes passthrough metrics to backends.
+type passThroughWriter struct {
+	numShards int
+	writers   []writer.Writer


can you make a new writer.Writer w/ the locking inside it? will make this simpler to read.

prateek · 2020-01-13T22:19:17Z

src/aggregator/server/m3msg/server.go

+		sp policy.StoragePolicy,
+		callback m3msg.Callbackable,
+	) {
+		// The type of a pass-through metric does not matter as it is written directly into m3db.


hm, doesn't the incoming metric have a type?

prateek · 2020-01-13T22:19:54Z

src/aggregator/server/m3msg/server.go

+			Value:     value,
+		}
+		metadata := metadata.TimedMetadata{
+			AggregationID: aggregation.MustCompressTypes(aggregation.Last),


make a const for this rather than do this at every invocation

prateek · 2020-01-13T22:20:49Z

src/aggregator/server/m3msg/server.go

+		}
+
+		if err := aggregator.AddPassThrough(metric, metadata); err != nil {
+			log.Info("[FAIL] to write pass-through metric",


can you remove these logs?

prateek · 2020-01-13T22:21:07Z

src/aggregator/server/m3msg/server.go

+
+		if err := aggregator.AddPassThrough(metric, metadata); err != nil {
+			log.Info("[FAIL] to write pass-through metric",
+				zap.String("metric", metric.String()),


nit: use zap.Stringer in cases like this

prateek · 2020-01-13T22:21:31Z

src/aggregator/server/m3msg/server.go

+			callback.Callback(m3msg.OnRetriableError)
+		}
+
+		if s != nil && s.Sample() {


should definitely remove these logs.

prateek · 2020-01-13T22:23:50Z

src/aggregator/server/m3msg/server.go

+	}
+}
+
+type addPassThroughError struct {


don't see any uses of this type, do you need it?

mway · 2020-01-29T17:38:25Z

src/aggregator/aggregator/handler/writer/sharded.go

+	var multiErr xerrors.MultiError
+	for i := 0; i < w.numShards; i++ {
+		w.locks[i].Lock()
+		multiErr = multiErr.Add(w.writers[i].Close())
+		w.locks[i].Unlock()
+	}
+
+	if multiErr.Empty() {
+		return nil
+	}
+
+	return fmt.Errorf("failed to close sharded writer: %v", multiErr.FinalError())


We should probably start to standardize on go.uber.org/multierr and github.com/pkg/errors as that's what we use everywhere else now.

This then becomes:

var err error for i := 0; i < w.numShards; i++ { w.locks[i].Lock() err = multierr.Append(err, w.writers[i].Close()) w.locks[i].Unlock() } return errors.WithMessage(err, "failed to close sharded writer")

talked offline, added errors.WithMessage

mway · 2020-01-29T17:42:57Z

src/aggregator/aggregator/handler/writer/sharded.go

+	w.Lock()
+	defer w.Unlock()
+
+	if w.closed {
+		return errShardedWriterClosed
+	}
+	w.closed = true


Thoughts on breaking reads of closed out into a helper to simplify mutex usage?

Hm need the lock for the duration of the functions in all three: Write(), Close() and Flush(). Don't see the formulation you're thinking of here.

src/aggregator/integration/options.go

mway · 2020-01-29T17:47:36Z

src/aggregator/server/m3msg/server.go

+		sp policy.StoragePolicy,
+		callback m3msg.Callbackable,
+	) {
+		// The type of a pass-through metric does not matter.


(just update this docstring)

mway · 2020-01-29T18:02:13Z

src/aggregator/aggregator/aggregator.go

+	agg.RLock()
+	if agg.state != aggregatorOpen {


Worth making this a helper?

mway · 2020-01-29T18:06:17Z

src/aggregator/aggregator/handler/writer/sharded.go

+	closed bool
+
+	numShards int
+	writers   []Writer


style: no spacing except for embeds/non-embeds

done for this instance but don't always agree: sometimes i break things into groups based on vars that are mutated & the locks that guard them.

src/aggregator/aggregator/handler/writer/sharded.go

mway · 2020-01-29T18:08:13Z

src/cmd/services/m3aggregator/config/aggregator.go

+	PassThroughTopicName *string `yaml:"passThroughTopicName"`
+
+	// The number of passthrough writers
+	NumPassThroughWriters *int `yaml:"numPassThroughWriters"`


Do these need to be nullable? Can we just use defaults + config population?

mway · 2020-01-29T18:12:19Z

src/cmd/services/m3aggregator/config/aggregator.go

+	for _, handler := range flushCfg.Handlers {
+		if handler.DynamicBackend != nil && c.PassThroughTopicName != nil {
+			handler.DynamicBackend.Producer.Writer.TopicName = *c.PassThroughTopicName
+		}
+	}


Hmm... not the most friendly code, esp for how simple it is. IMO further datapoints to not use config pointers if we can avoid it.

talked offline: going to leave as is because the changes are pretty pervasive to make handler not be a pointer type.

mway · 2020-01-29T18:13:20Z

src/cmd/services/m3aggregator/serve/serve.go

@@ -56,6 +63,12 @@ func Serve(
 	defer httpServer.Close()
 	log.Infof("http server: listening on %s", httpAddr)

+	if err := m3msgServer.ListenAndServe(); err != nil {
+		return fmt.Errorf("could not start m3msg server at %s: %v", m3msgAddr, err)


errors.WithMessage

codecov · 2020-02-06T00:24:57Z

Codecov Report

Merging #2105 into master will decrease coverage by 1.1%.
The diff coverage is 46.9%.

@@           Coverage Diff            @@
##           master   #2105     +/-   ##
========================================
- Coverage    65.5%   64.4%   -1.2%     
========================================
  Files         961    1008     +47     
  Lines       85488   86916   +1428     
========================================
- Hits        56021   55983     -38     
- Misses      25367   27017   +1650     
+ Partials     4100    3916    -184

Flag	Coverage Δ
#aggregator	`83.3% <47.2%> (-16.7%)`	⬇️
#cluster	`72.5% <ø> (-4.9%)`	⬇️
#collector	`71.4% <ø> (+40.7%)`	⬆️
#dbnode	`77.5% <ø> (-22.5%)`	⬇️
#m3em	`58.1% <ø> (-41.9%)`	⬇️
#m3ninx	`65.8% <ø> (-34.2%)`	⬇️
#m3nsch	`51.1% <ø> (-26.9%)`	⬇️
#metrics	`17.6% <ø> (-82.4%)`	⬇️
#msg	`74.9% <0%> (-25.1%)`	⬇️
#query	`44% <ø> (-56%)`	⬇️
#x	`71.9% <ø> (-28.1%)`	⬇️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 6db9eee...1cbd516. Read the comment docs.

mway · 2020-02-10T20:07:20Z

src/aggregator/server/m3msg/server.go

+		sp policy.StoragePolicy,
+		callback m3msg.Callbackable,
+	) {
+		// The type of a pass-through metric does not matter.


(just update this docstring)

mway · 2020-02-10T20:09:07Z

src/cmd/services/m3aggregator/serve/serve.go

@@ -27,6 +27,13 @@ import (
 	httpserver "github.com/m3db/m3/src/aggregator/server/http"
 	rawtcpserver "github.com/m3db/m3/src/aggregator/server/rawtcp"
 	"github.com/m3db/m3/src/x/instrument"
+	"github.com/m3db/m3/src/x/server"
+


nit: remove space

mway · 2020-02-10T20:09:25Z

src/cmd/services/m3aggregator/main/main.go

 	"github.com/m3db/m3/src/cmd/services/m3aggregator/config"
 	"github.com/m3db/m3/src/cmd/services/m3aggregator/serve"
 	xconfig "github.com/m3db/m3/src/x/config"
 	"github.com/m3db/m3/src/x/config/configflag"
 	"github.com/m3db/m3/src/x/etcd"
 	"github.com/m3db/m3/src/x/instrument"
+	"github.com/m3db/m3/src/x/server"



nit: remove this space (even though you didn't add it)

CLAassistant · 2021-01-27T04:37:36Z

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
1 out of 2 committers have signed the CLA.

✅ prateek
❌ Siyu Yang

Siyu Yang seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

fishie9 requested a review from prateek January 13, 2020 21:39