Add M3 reporter backend to tally (uber-go#24)

.gitignore

@@ -28,4 +28,6 @@ _testmain.go
 *.log
 
 .DS_Store
+node_modules/
+
 

.travis.yml

@@ -12,6 +12,7 @@ cache:
   directories:
     - vendor
 install:
+  - npm i uber-licence
   - make dependencies
 script:
   - make test

Makefile

@@ -2,7 +2,8 @@ export GO15VENDOREXPERIMENT=1
 
 BENCH_FLAGS ?= -cpuprofile=cpu.pprof -memprofile=mem.pprof -benchmem
 PKGS ?= $(shell glide novendor)
-PKG_FILES ?= *.go example/*.go
+PKG_FILES ?= *.go example/*.go m3
+LINT_IGNORE = m3/thrift
 
 # The linting tools evolve with each Go version, so run them only on the latest
 # stable release.
@@ -37,15 +38,15 @@ lint:
 ifdef SHOULD_LINT
 	@rm -rf lint.log
 	@echo "Checking formatting..."
-	@gofmt -d -s $(PKG_FILES) 2>&1 | tee lint.log
+	@gofmt -d -s $(PKG_FILES) 2>&1 | grep -v $(LINT_IGNORE) | tee lint.log
 	@echo "Installing test dependencies for vet..."
 	@go test -i $(PKGS)
 	@echo "Checking vet..."
-	@$(foreach dir,$(PKG_FILES),go tool vet $(dir) 2>&1 | tee -a lint.log;)
+	@$(foreach dir,$(PKG_FILES),go tool vet $(dir) 2>&1 | grep -v $(LINT_IGNORE) | tee -a lint.log;)
 	@echo "Checking lint..."
-	@$(foreach dir,$(PKGS),golint $(dir) 2>&1 | tee -a lint.log;)
+	@$(foreach dir,$(PKGS),golint $(dir) 2>&1 | grep -v $(LINT_IGNORE) | tee -a lint.log;)
 	@echo "Checking for unresolved FIXMEs..."
-	@git grep -i fixme | grep -v -e vendor -e Makefile | tee -a lint.log
+	@git grep -i fixme | grep -v -e vendor -e Makefile | grep -v $(LINT_IGNORE) | tee -a lint.log
 	@echo "Checking for license headers..."
 	@./check_license.sh | tee -a lint.log
 	@[ ! -s lint.log ]
@@ -55,11 +56,11 @@ endif
 
 .PHONY: test
 test:
-	go test -race $(PKGS)
+	go test -race -v $(PKGS)
 
 .PHONY: cover
 cover:
-	go test -cover -coverprofile cover.out -race .
+	go test -cover -coverprofile cover.out -race -v $(PKGS)
 
 .PHONY: coveralls
 coveralls:

check_license.sh

@@ -1,15 +1,4 @@
 #!/bin/bash
 
-text=`head -1 LICENSE`
-
-ERROR_COUNT=0
-while read file
-do
-    head -1 ${file} | grep -q "${text}"
-    if [ $? -ne 0 ]; then
-        echo "$file is missing license header."
-        ERROR_COUNT+=1
-    fi
-done < <(git ls-files "\.go")
-
-exit $ERROR_COUNT
+./node_modules/.bin/uber-licence --version || npm i uber-licence@latest
+./node_modules/.bin/uber-licence --dry --file "*.go"

example/main.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2016 Uber Technologies, Inc.
+// Copyright (c) 2017 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

glide.yaml

@@ -11,6 +11,12 @@ import:
   version: ^0.8.0
   subpackages:
   - prometheus
+- package: github.com/apache/thrift
+  version: 9549b25c77587b29be4e0b5c258221a4ed85d37a
+  subpackages:
+  - lib/go/thrift
+- package: github.com/uber-go/atomic
+  version: e682c1008ac17bf26d2e4b5ad6cdd08520ed0b22
 testImport:
 - package: github.com/axw/gocov
   version: 54b98cfcac0c63fb3f9bd8e7ad241b724d4e985b

m3/config.go

@@ -0,0 +1,65 @@
+// Copyright (c) 2017 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 m3
+
+// Configuration is a configuration for a M3 reporter.
+type Configuration struct {
+	// HostPort is the host and port of the M3 server.
+	HostPort string `yaml:"hostPort" validate:"nonzero"`
+
+	// HostPorts are the host and port of the M3 server.
+	HostPorts []string `yaml:"hostPorts"`
+
+	// Service is the service tag to that this client emits.
+	Service string `yaml:"service" validate:"nonzero"`
+
+	// Env is the env tag to use that this client emits.
+	Env string `yaml:"env" validate:"nonzero"`
+
+	// CommonTags are tags that are common for all metrics this client emits.
+	CommonTags map[string]string `yaml:"tags" `
+
+	// Queue is the maximum metric queue size of client.
+	Queue int `yaml:"queue"`
+
+	// PacketSize is the maximum packet size for a batch of metrics.
+	PacketSize int32 `yaml:"packetSize"`
+
+	// IncludeHost is whether or not to include host tag.
+	IncludeHost bool `yaml:"includeHost"`
+}
+
+// NewReporter creates a new M3 reporter from this configuration.
+func (c Configuration) NewReporter() (Reporter, error) {
+	hostPorts := c.HostPorts
+	if len(hostPorts) == 0 {
+		hostPorts = []string{c.HostPort}
+	}
+	return NewReporter(Options{
+		HostPorts:          hostPorts,
+		Service:            c.Service,
+		Env:                c.Env,
+		CommonTags:         c.CommonTags,
+		MaxQueueSize:       c.Queue,
+		MaxPacketSizeBytes: c.PacketSize,
+		IncludeHost:        c.IncludeHost,
+	})
+}

m3/config_test.go

@@ -0,0 +1,62 @@
+// Copyright (c) 2017 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 m3
+
+import (
+	"testing"
+
+	"github.com/uber-go/tally/m3/thriftudp"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigSimple(t *testing.T) {
+	c := Configuration{
+		HostPort: "127.0.0.1:9052",
+		Service:  "my-service",
+		Env:      "test",
+	}
+	r, err := c.NewReporter()
+	require.NoError(t, err)
+
+	reporter := r.(*reporter)
+	_, ok := reporter.client.Transport.(*thriftudp.TUDPTransport)
+	assert.True(t, ok)
+	assert.True(t, tagEquals(reporter.commonTags, "service", "my-service"))
+	assert.True(t, tagEquals(reporter.commonTags, "env", "test"))
+}
+
+func TestConfigMulti(t *testing.T) {
+	c := Configuration{
+		HostPorts: []string{"127.0.0.1:9052", "127.0.0.1:9062"},
+		Service:   "my-service",
+		Env:       "test",
+	}
+	r, err := c.NewReporter()
+	require.NoError(t, err)
+
+	reporter := r.(*reporter)
+	_, ok := reporter.client.Transport.(*thriftudp.TMultiUDPTransport)
+	assert.True(t, ok)
+	assert.True(t, tagEquals(reporter.commonTags, "service", "my-service"))
+	assert.True(t, tagEquals(reporter.commonTags, "env", "test"))
+}

m3/customtransports/buffered_read_transport.go

@@ -0,0 +1,79 @@
+// Copyright (c) 2017 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 customtransport
+
+import (
+	"bytes"
+
+	"github.com/apache/thrift/lib/go/thrift"
+)
+
+// TBufferedReadTransport is a thrift.TTransport that reads from a buffer
+type TBufferedReadTransport struct {
+	readBuf *bytes.Buffer
+}
+
+// NewTBufferedReadTransport creates a buffer backed TTransport
+func NewTBufferedReadTransport(readBuf *bytes.Buffer) (*TBufferedReadTransport, error) {
+	return &TBufferedReadTransport{readBuf: readBuf}, nil
+}
+
+// IsOpen does nothing as transport is not maintaining the connection
+// Required to maintain thrift.TTransport interface
+func (p *TBufferedReadTransport) IsOpen() bool {
+	return true
+}
+
+// Open does nothing as transport is not maintaining the connection
+// Required to maintain thrift.TTransport interface
+func (p *TBufferedReadTransport) Open() error {
+	return nil
+}
+
+// Close does nothing as transport is not maintaining the connection
+// Required to maintain thrift.TTransport interface
+func (p *TBufferedReadTransport) Close() error {
+	return nil
+}
+
+// Read reads bytes from the local buffer and puts them in the specified buf
+func (p *TBufferedReadTransport) Read(buf []byte) (int, error) {
+	in, err := p.readBuf.Read(buf)
+	return in, thrift.NewTTransportExceptionFromError(err)
+}
+
+// RemainingBytes returns the number of bytes left to be read from the readBuf
+func (p *TBufferedReadTransport) RemainingBytes() uint64 {
+	return uint64(p.readBuf.Len())
+}
+
+// Write writes bytes into the read buffer
+// Required to maintain thrift.TTransport interface
+func (p *TBufferedReadTransport) Write(buf []byte) (int, error) {
+	p.readBuf = bytes.NewBuffer(buf)
+	return len(buf), nil
+}
+
+// Flush does nothing as udp server does not write responses back
+// Required to maintain thrift.TTransport interface
+func (p *TBufferedReadTransport) Flush() error {
+	return nil
+}