Add storage based ingester to m3coordinator #1038

cw9 · 2018-10-08T16:57:04Z

No description provided.

codecov · 2018-10-08T18:25:25Z

Codecov Report

Merging #1038 into master will decrease coverage by 0.03%.
The diff coverage is 36.36%.

@@            Coverage Diff             @@
##           master    #1038      +/-   ##
==========================================
- Coverage   77.88%   77.85%   -0.04%     
==========================================
  Files         530      532       +2     
  Lines       45304    45361      +57     
==========================================
+ Hits        35287    35316      +29     
- Misses       7799     7819      +20     
- Partials     2218     2226       +8

Flag	Coverage Δ
#aggregator	`81.67% <ø> (-0.03%)`	⬇️
#collector	`96.47% <ø> (ø)`	⬆️
#dbnode	`81.3% <ø> (-0.04%)`	⬇️
#m3em	`73.21% <ø> (ø)`	⬆️
#m3ninx	`75.33% <ø> (+0.07%)`	⬆️
#m3nsch	`51.19% <ø> (ø)`	⬆️
#query	`64.1% <0%> (-0.13%)`	⬇️
#x	`76.2% <100%> (+0.98%)`	⬆️

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 440b416...670939d. Read the comment docs.

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

@@ -0,0 +1,65 @@
+package ingest


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

@@ -0,0 +1,175 @@
+package ingest


robskillington · 2018-10-08T19:59:05Z

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

+
+// NewIngester creates an ingester.
+func NewIngester(
+	opts *Options,


Perhaps we should make this non-pointer type? Options should be required it looks like so no need to make pointer type.

robskillington · 2018-10-08T20:34:36Z

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

+// Ingest ingests a metric asynchronously with callback.
+func (i *Ingester) Ingest(
+	id []byte,
+	metricTimeNanos int64,


Should we perhaps just make this time.Time considering it's converted to that later anyhow? Less chance of getting the precision wrong if so.

robskillington · 2018-10-08T20:36:28Z

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

+	return op.s.Write(
+		// NB: No timeout is needed for this as the m3db client has a timeout
+		// configured to it.
+		context.Background(),


Hm is it worth plumbing a ctx all the way from the top to here? ctx.Background() is a bit untrue, the context really should come from the ingester server.

I could, but then the server also does not have anything to add to the context. This is mainly to fulfill the storage interface, I'll just make a context whenever a new ingestOp is made like statsdex.

robskillington · 2018-10-08T20:40:54Z

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

+	}
+	op.resetDataPoints()
+	op.q.Raw = string(op.id)
+	op.q.Unit = op.sp.Resolution().Precision


I just checked how Precision gets resolved for say a storage policy of 1m:30d and it seems like it's going to set Unit of xtime.Minute which won't be compatible with M3DB (which at most supports a unit/precision of time.Second).

I'll just use xtime.Millisecond then, seems that's the default value used in several places in query

Updating the usage in flush_handler.go as well

robskillington · 2018-10-08T20:41:56Z

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

+	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: name, Value: value})


This is unsafe to do, this is the bug I just fixed where you can't hold onto the byte refs from the tag iterator since it reuses them.

robskillington · 2018-10-08T20:42:57Z

src/cmd/services/m3query/config/config.go

+	Ingester ingest.Configuration `yaml:"ingester"`
+
+	// M3msg is the configuration for m3msg server.
+	M3msg m3msg.Configuration `yaml:"m3msg"`


Technically this should probably be (due to camel casing convention):

M3Msg m3msg.Configuration `yaml:"m3msg"`

robskillington · 2018-10-08T22:01:51Z

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

@@ -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,


This will have pretty poor compression for things that can be written in with second precision.

Could we please do at least something here where we try to find the best? i.e.

switch { case time.Duration(nanos) % time.Second == 0: unit = xtime.Second case time.Duration(nanos) % time.Millisecond == 0: unit = xtime.Millisecond case time.Duration(nanos) % time.Microsecond == 0: unit = xtime.Microsecond default: unit = xtime.Nanosecond }

Can you update flush_handler.go to use the new method you created? common.SanitizeUnitForM3DB(mp.StoragePolicy.Resolution().Precision)

yeah, forgot about this

robskillington · 2018-10-08T22:03:25Z

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

+	metricTime time.Time,
+	value float64,
+	sp policy.StoragePolicy,
+	callback *m3msg.RefCountedCallback,


Can we add ctx context.Context as the first argument? (lint forces some context to always be the first arg)

We don't want to use the TODO context each time.

hmm, the m3msg server will only be able to pass a TODO/Background context each time though, although I can reuse one context there, is it what you wanted?

Yes I think that's ideal, in the future if we have a per-op context then we'll be able to use that. Otherwise here it'll be a single one that can never be specialized. We should at least let the callers into this layer be able to specify their own.

robskillington · 2018-10-08T22:03:41Z

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

+				it: serialize.NewMetricTagsIterator(tagDecoder, nil),
+				p:  p,
+				m:  m,
+				c:  context.TODO(),


This can just be nil if it gets set on each ingest op in .Ingest(...).

robskillington · 2018-10-08T22:10:24Z

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

+		return err
+	}
+	op.resetDataPoints()
+	op.q.Raw = string(op.id)


This actually doesn't need to be set, while you're here can you remove Raw from the storage.WriteQuery type? I just checked references and it doesn't get used anywhere except the String() method on storage.WriteQuery (which we can replace with q.Tags.String() or similar instead of q.Raw).

robskillington · 2018-10-08T23:30:59Z

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

+func (cfg Configuration) newOptions(
+	appender storage.Appender,
+	instrumentOptions instrument.Options,
+) (*Options, error) {


Since you need to derefence this pointer later now for use in the constructor, perhaps make this return type (Options, error)?

yeah, was trying to avoid returning Options{}, err, I can change that

robskillington · 2018-10-08T23:33:06Z

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

+		return err
+	}
+	op.resetDataPoints()
+	op.q.Unit = common.SanitizeUnitForM3DB(op.sp.Resolution().Precision)


robskillington · 2018-10-08T23:42:12Z

src/cmd/services/m3coordinator/server/m3msg/handler.go

@@ -32,6 +34,8 @@ import (
 	"github.com/uber-go/tally"
 )

+var ctx = context.TODO()


Instead of making this global, can you give one to each handler, then pass that to each per consumer handler?

Also it makes more sense to derive from context.Background() rather than context.TODO() since we want to signify this was from a background listen server, rather than intending to replace it later.

robskillington · 2018-10-08T23:50:26Z

src/x/common/unit.go

+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package common


minor nit: Can we rename this to convert? We could rename SanitizeUnitForM3DB to something like UnitForM3DB(unit xtime.Unit) xtime.Unit and start putting other conversion functions in here.

robskillington

LGTM

Add a storage based ingester to m3coordinator

85c233c

cw9 force-pushed the chao/ingest branch from dc9c0a4 to 85c233c Compare October 8, 2018 18:24

robskillington reviewed Oct 8, 2018

View reviewed changes

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

@@ -0,0 +1,65 @@

package ingest

This comment was marked as resolved.

Sign in to view

robskillington reviewed Oct 8, 2018

View reviewed changes

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

@@ -0,0 +1,175 @@

package ingest

This comment was marked as resolved.

Sign in to view

add license

4fc8a72

robskillington reviewed Oct 8, 2018

View reviewed changes

Address comments

0370602

robskillington reviewed Oct 8, 2018

View reviewed changes

Adderss comments

43f75c0

robskillington reviewed Oct 8, 2018

View reviewed changes

Address comments

9cef1e5

robskillington approved these changes Oct 9, 2018

View reviewed changes

cw9 and others added 2 commits October 9, 2018 10:48

Merge branch 'master' into chao/ingest

c915fc3

glide up

670939d

cw9 merged commit 2ffc763 into master Oct 9, 2018

cw9 deleted the chao/ingest branch October 9, 2018 15:49

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add storage based ingester to m3coordinator #1038

Add storage based ingester to m3coordinator #1038

cw9 commented Oct 8, 2018

codecov bot commented Oct 8, 2018 •

edited

Loading

This comment was marked as resolved.

This comment was marked as resolved.

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

cw9 Oct 8, 2018

robskillington Oct 8, 2018 •

edited

Loading

cw9 Oct 8, 2018

cw9 Oct 8, 2018

robskillington Oct 8, 2018

robskillington Oct 8, 2018

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018

cw9 Oct 8, 2018

robskillington Oct 8, 2018

cw9 Oct 8, 2018

robskillington Oct 8, 2018

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018

cw9 Oct 8, 2018

robskillington Oct 8, 2018

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington left a comment

Add storage based ingester to m3coordinator #1038

Add storage based ingester to m3coordinator #1038

Conversation

cw9 commented Oct 8, 2018

codecov bot commented Oct 8, 2018 • edited Loading

Codecov Report

This comment was marked as resolved.

This comment was marked as resolved.

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

robskillington Oct 8, 2018 • edited Loading

Choose a reason for hiding this comment

robskillington left a comment

Choose a reason for hiding this comment

codecov bot commented Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading

robskillington Oct 8, 2018 •

edited

Loading