Merge pull request #156 from yandex/arcadia

Arcadia

cli/cli.go

@@ -24,7 +24,7 @@ import (
 	"go.uber.org/zap/zapcore"
 )
 
-const Version = "0.4.0"
+const Version = "0.5.0"
 const defaultConfigFile = "load"
 const stdinConfigSelector = "-"
 

components/phttp/base.go

@@ -12,6 +12,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"net/http/httptrace"
 	"net/http/httputil"
 	"net/url"
 
@@ -26,8 +27,9 @@ const (
 )
 
 type BaseGunConfig struct {
-	AutoTag AutoTagConfig `config:"auto-tag"`
-	AnswLog AnswLogConfig `config:"answlog"`
+	AutoTag   AutoTagConfig   `config:"auto-tag"`
+	AnswLog   AnswLogConfig   `config:"answlog"`
+	HTTPTrace HTTPTraceConfig `config:"httptrace"`
 }
 
 // AutoTagConfig configure automatic tags generation based on ammo URI. First AutoTag URI path elements becomes tag.
@@ -44,6 +46,11 @@ type AnswLogConfig struct {
 	Filter  string `config:"filter" valid:"oneof=all warning error"`
 }
 
+type HTTPTraceConfig struct {
+	DumpEnabled  bool `config:"dump"`
+	TraceEnabled bool `config:"trace"`
+}
+
 func DefaultBaseGunConfig() BaseGunConfig {
 	return BaseGunConfig{
 		AutoTagConfig{
@@ -56,6 +63,10 @@ func DefaultBaseGunConfig() BaseGunConfig {
 			Path:    "answ.log",
 			Filter:  "error",
 		},
+		HTTPTraceConfig{
+			DumpEnabled:  false,
+			TraceEnabled: false,
+		},
 	}
 }
 
@@ -136,8 +147,41 @@ func (b *BaseGun) Shoot(ammo Ammo) {
 		err = errors.WithStack(err)
 	}()
 
+	var timings *TraceTimings
+	if b.Config.HTTPTrace.TraceEnabled {
+		var clientTracer *httptrace.ClientTrace
+		clientTracer, timings = createHTTPTrace()
+		req = req.WithContext(httptrace.WithClientTrace(req.Context(), clientTracer))
+	}
+	if b.Config.HTTPTrace.DumpEnabled {
+		requestDump, err := httputil.DumpRequest(req, true)
+		if err != nil {
+			b.Log.Error("DumpRequest error", zap.Error(err))
+		} else {
+			sample.SetRequestBytes(len(requestDump))
+		}
+	}
 	var res *http.Response
 	res, err = b.Do(req)
+
+	if b.Config.HTTPTrace.TraceEnabled && timings != nil {
+		sample.SetReceiveTime(timings.GetReceiveTime())
+	}
+
+	if b.Config.HTTPTrace.DumpEnabled && res != nil {
+		responseDump, err := httputil.DumpResponse(res, true)
+		if err != nil {
+			b.Log.Error("DumpResponse error", zap.Error(err))
+		} else {
+			sample.SetResponseBytes(len(responseDump))
+		}
+	}
+	if b.Config.HTTPTrace.TraceEnabled && timings != nil {
+		sample.SetConnectTime(timings.GetConnectTime())
+		sample.SetSendTime(timings.GetSendTime())
+		sample.SetLatency(timings.GetLatency())
+	}
+
 	if err != nil {
 		b.Log.Warn("Request fail", zap.Error(err))
 		return

components/phttp/client.go

@@ -13,7 +13,6 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
-	"github.com/yandex/pandora/core/config"
 	"github.com/yandex/pandora/lib/netutil"
 	"go.uber.org/zap"
 	"golang.org/x/net/http2"
@@ -64,8 +63,12 @@ func DefaultDialerConfig() DialerConfig {
 }
 
 func NewDialer(conf DialerConfig) netutil.Dialer {
-	d := &net.Dialer{}
-	config.Map(d, conf)
+	d := &net.Dialer{
+		Timeout:       conf.Timeout,
+		DualStack:     conf.DualStack,
+		FallbackDelay: conf.FallbackDelay,
+		KeepAlive:     conf.KeepAlive,
+	}
 	if !conf.DNSCache {
 		return d
 	}
@@ -96,12 +99,20 @@ func DefaultTransportConfig() TransportConfig {
 }
 
 func NewTransport(conf TransportConfig, dial netutil.DialerFunc) *http.Transport {
-	tr := &http.Transport{}
+	tr := &http.Transport{
+		TLSHandshakeTimeout:   conf.TLSHandshakeTimeout,
+		DisableKeepAlives:     conf.DisableKeepAlives,
+		DisableCompression:    conf.DisableCompression,
+		MaxIdleConns:          conf.MaxIdleConns,
+		MaxIdleConnsPerHost:   conf.MaxIdleConnsPerHost,
+		IdleConnTimeout:       conf.IdleConnTimeout,
+		ResponseHeaderTimeout: conf.ResponseHeaderTimeout,
+		ExpectContinueTimeout: conf.ExpectContinueTimeout,
+	}
 	tr.TLSClientConfig = &tls.Config{
 		InsecureSkipVerify: true,                 // We should not spend time for this stuff.
 		NextProtos:         []string{"http/1.1"}, // Disable HTTP/2. Use HTTP/2 transport explicitly, if needed.
 	}
-	config.Map(tr, conf)
 	tr.DialContext = dial
 	return tr
 }

components/phttp/trace.go

@@ -0,0 +1,65 @@
+package phttp
+
+import (
+	"net/http/httptrace"
+	"time"
+)
+
+type TraceTimings struct {
+	GotConnTime          time.Time
+	GetConnTime          time.Time
+	DNSStartTime         time.Time
+	DNSDoneTime          time.Time
+	ConnectDoneTime      time.Time
+	ConnectStartTime     time.Time
+	WroteRequestTime     time.Time
+	GotFirstResponseByte time.Time
+}
+
+func (t *TraceTimings) GetReceiveTime() time.Duration {
+	return time.Since(t.GotFirstResponseByte)
+}
+
+func (t *TraceTimings) GetConnectTime() time.Duration {
+	return t.GotConnTime.Sub(t.GetConnTime)
+}
+
+func (t *TraceTimings) GetSendTime() time.Duration {
+	return t.WroteRequestTime.Sub(t.GotConnTime)
+}
+
+func (t *TraceTimings) GetLatency() time.Duration {
+	return t.GotFirstResponseByte.Sub(t.WroteRequestTime)
+}
+
+func createHTTPTrace() (*httptrace.ClientTrace, *TraceTimings) {
+	timings := &TraceTimings{}
+	tracer := &httptrace.ClientTrace{
+		GetConn: func(_ string) {
+			timings.GetConnTime = time.Now()
+		},
+		GotConn: func(_ httptrace.GotConnInfo) {
+			timings.GotConnTime = time.Now()
+		},
+		DNSStart: func(_ httptrace.DNSStartInfo) {
+			timings.DNSStartTime = time.Now()
+		},
+		DNSDone: func(info httptrace.DNSDoneInfo) {
+			timings.DNSDoneTime = time.Now()
+		},
+		ConnectStart: func(network, addr string) {
+			timings.ConnectStartTime = time.Now()
+		},
+		ConnectDone: func(network, addr string, err error) {
+			timings.ConnectDoneTime = time.Now()
+		},
+		WroteRequest: func(wr httptrace.WroteRequestInfo) {
+			timings.WroteRequestTime = time.Now()
+		},
+		GotFirstResponseByte: func() {
+			timings.GotFirstResponseByte = time.Now()
+		},
+	}
+
+	return tracer, timings
+}

core/aggregator/jsonlines.go

@@ -11,7 +11,6 @@ import (
 
 	jsoniter "github.com/json-iterator/go"
 	"github.com/yandex/pandora/core"
-	"github.com/yandex/pandora/core/config"
 	"github.com/yandex/pandora/core/coreutil"
 	"github.com/yandex/pandora/lib/ioutil2"
 )
@@ -51,8 +50,11 @@ func NewJSONLinesAggregator(conf JSONLineAggregatorConfig) core.Aggregator {
 }
 
 func NewJSONEncoder(w io.Writer, conf JSONLineEncoderConfig) SampleEncoder {
-	var apiConfig jsoniter.Config
-	config.Map(&apiConfig, conf.JSONIterConfig)
+	apiConfig := jsoniter.Config{
+		SortMapKeys:             conf.JSONIterConfig.SortMapKeys,
+		MarshalFloatWith6Digits: conf.JSONIterConfig.MarshalFloatWith6Digits,
+	}
+
 	api := apiConfig.Froze()
 	// NOTE(skipor): internal buffering is not working really. Don't know why
 	// OPTIMIZE(skipor): don't wrap into buffer, if already ioutil2.ByteWriter

core/aggregator/netsample/sample.go

@@ -103,15 +103,27 @@ func (s *Sample) SetUserNet(code int) {
 	s.set(keyErrno, code)
 }
 
+func (s *Sample) SetConnectTime(d time.Duration) {
+	s.setDuration(keyConnectMicro, d)
+}
+
+func (s *Sample) SetSendTime(d time.Duration) {
+	s.setDuration(keySendMicro, d)
+}
+
 func (s *Sample) SetLatency(d time.Duration) {
 	s.setDuration(keyLatencyMicro, d)
 }
 
+func (s *Sample) SetReceiveTime(d time.Duration) {
+	s.setDuration(keyReceiveMicro, d)
+}
+
 func (s *Sample) SetRequestBytes(b int) {
 	s.set(keyRequestBytes, b)
 }
 
-func (s *Sample) SetResponceBytes(b int) {
+func (s *Sample) SetResponseBytes(b int) {
 	s.set(keyResponseBytes, b)
 }
 

core/aggregator/netsample/sample_test.go

@@ -68,7 +68,7 @@ func TestCustomSets(t *testing.T) {
 	s.SetRequestBytes(reqBytes)
 
 	respBytes := 8
-	s.SetResponceBytes(respBytes)
+	s.SetResponseBytes(respBytes)
 
 	expectedTimeStamp := fmt.Sprintf("%v.%3.f",
 		s.timeStamp.Unix(),

core/config/config.go

@@ -6,7 +6,6 @@
 package config
 
 import (
-	"github.com/fatih/structs"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 )
@@ -39,18 +38,36 @@ func DecodeAndValidate(conf interface{}, result interface{}) error {
 // in such case you can from this subset of fields struct Single, decode config
 // into it, and map it on Multi.
 func Map(dst, src interface{}) {
-	conf := &mapstructure.DecoderConfig{
+	// dst and src conf for compatibility with old fatih/structs.
+	// src map from "map:" tags -> tmp -> map to "mapstructure:" tags in dst
+	dstConf := &mapstructure.DecoderConfig{
 		ErrorUnused: true,
 		ZeroFields:  true,
 		Result:      dst,
 	}
-	d, err := mapstructure.NewDecoder(conf)
+	d, err := mapstructure.NewDecoder(dstConf)
 	if err != nil {
 		panic(err)
 	}
-	s := structs.New(src)
-	s.TagName = "map"
-	err = d.Decode(s.Map())
+
+	tmp := make(map[string]interface{})
+	srcConf := &mapstructure.DecoderConfig{
+		ErrorUnused: true,
+		ZeroFields:  true,
+		Result:      &tmp,
+		TagName:     "map",
+	}
+	s, err := mapstructure.NewDecoder(srcConf)
+	if err != nil {
+		panic(err)
+	}
+
+	err = s.Decode(src)
+	if err != nil {
+		panic(err)
+	}
+
+	err = d.Decode(tmp)
 	if err != nil {
 		panic(err)
 	}

core/config/config_test.go

@@ -186,7 +186,6 @@ func TestMapTagged(t *testing.T) {
 	Map(n, &M{SomeOtherFieldName: MultiStrings{A: "a"}})
 	assert.Equal(t, &N{A: "a", MultiStrings: MultiStrings{A: "a"}}, n)
 }
-
 func TestDeltaUpdate(t *testing.T) {
 	var l2 Level2
 	err := Decode(M{
@@ -242,22 +241,26 @@ func TestConfigEnvVarReplacement(t *testing.T) {
 	t.Setenv("VAR_2", "value2")
 	t.Setenv("INT_VAR_3", "15")
 	t.Setenv("IP_SEQ", "1.2")
+	t.Setenv("DURATION", "30s")
 	var l1 struct {
 		Val1 string
 		Val2 string
 		Val3 int
 		Val4 net.IP
+		Val5 time.Duration
 	}
 
 	err := Decode(M{
 		"val1": "aa-${ENV_VAR_1}",
 		"val2": "${ENV:VAR_2}",
 		"val3": "${INT_VAR_3}",
 		"val4": "1.1.${ENV:IP_SEQ}",
+		"val5": "${DURATION}",
 	}, &l1)
 	assert.NoError(t, err)
 	assert.Equal(t, "aa-value1", l1.Val1)
 	assert.Equal(t, "value2", l1.Val2)
 	assert.Equal(t, 15, l1.Val3)
 	assert.Equal(t, net.IPv4(1, 1, 1, 2), l1.Val4)
+	assert.Equal(t, 30*time.Second, l1.Val5)
 }

core/import/import.go

@@ -30,7 +30,7 @@ const (
 	compositeScheduleKey = "composite"
 )
 
-//getter for fs to avoid afero dependency in custom guns
+// getter for fs to avoid afero dependency in custom guns
 func GetFs() afero.Fs {
 	return afero.NewOsFs()
 }

go.mod

@@ -7,7 +7,6 @@ require (
 	github.com/c2h5oh/datasize v0.0.0-20171227191756-4eba002a5eae
 	github.com/facebookgo/stack v0.0.0-20160209184415-751773369052
 	github.com/facebookgo/stackerr v0.0.0-20150612192056-c2fcf88613f4
-	github.com/fatih/structs v1.0.0
 	github.com/ghodss/yaml v1.0.0
 	github.com/hashicorp/go-multierror v0.0.0-20171204182908-b7773ae21874
 	github.com/jhump/protoreflect v1.10.1
@@ -29,20 +28,21 @@ require (
 
 require (
 	github.com/davecgh/go-spew v1.1.0 // indirect
-	github.com/fsnotify/fsnotify v1.4.7 // indirect
+	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/go-playground/locales v0.11.2 // indirect
 	github.com/go-playground/universal-translator v0.16.0 // indirect
 	github.com/golang/protobuf v1.4.3 // indirect
 	github.com/hashicorp/errwrap v0.0.0-20141028054710-7554cd9344ce // indirect
 	github.com/hashicorp/hcl v0.0.0-20171017181929-23c074d0eceb // indirect
+	github.com/magiconair/properties v1.8.6 // indirect
 	github.com/pelletier/go-toml v1.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/spf13/cast v1.2.0 // indirect
 	github.com/spf13/jwalterweatherman v0.0.0-20180109140146-7c0cea34c8ec // indirect
 	github.com/spf13/pflag v1.0.0 // indirect
 	github.com/stretchr/objx v0.1.0 // indirect
 	go.uber.org/multierr v1.1.0 // indirect
-	golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd // indirect
+	golang.org/x/sys v0.0.0-20220926163933-8cfa568d3c25 // indirect
 	golang.org/x/text v0.3.0 // indirect
 	google.golang.org/genproto v0.0.0-20200526211855-cb27e3aa2013 // indirect
 	google.golang.org/protobuf v1.25.1-0.20200805231151-a709e31e5d12 // indirect