Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use dispatch method of statsd Package

Best K6 code snippet using statsd.dispatch

output.go

Source:output.go

...14   The various Output Pool writers objects/functions15*/16package cadent17import (18	"cadent/server/dispatch"19	"cadent/server/netpool"20	"cadent/server/schemas/repr"21	"cadent/server/stats"22	"fmt"23	"net"24	"net/url"25	"time"26)27// OutMessageWriter interface for output writers28type OutMessageWriter interface {29	Write(*OutputMessage) error30}31// PoolWriter a writer that does uses a pool of outgoing sockets for sending32type PoolWriter struct{}33// Write send the message to the backend pool sockets34func (p *PoolWriter) Write(j *OutputMessage) error {35	// to send stat lines via a pool of connections36	// rather then one socket per stat37	defer stats.StatsdNanoTimeFunc("worker.process-time-ns", time.Now())38	var outsrv netpool.NetpoolInterface39	var ok bool40	// lock out Outpool map41	if j == nil {42		return nil43	}44	make_pool := func() int {45		j.server.poolmu.Lock()46		defer j.server.poolmu.Unlock()47		if outsrv, ok = j.server.Outpool[j.outserver]; ok {48			ok = true49			return 050		} else {51			mURL, err := url.Parse(j.outserver)52			if err != nil {53				stats.StatsdClient.Incr("failed.bad-url", 1)54				j.server.FailSendCount.Up(1)55				j.server.log.Error("Error sending to backend Invalid URL `%s` %s", j.outserver, err)56				return 2 //cannot retry this57			}58			if len(j.server.Outpool) == 0 {59				j.server.Outpool = make(map[string]netpool.NetpoolInterface)60			}61			if j.server.SendingConnectionMethod == "bufferedpool" {62				outsrv = netpool.NewBufferedNetpool(mURL.Scheme, mURL.Host+mURL.Path, j.server.WriteBufferPoolSize)63			} else {64				outsrv = netpool.NewNetpool(mURL.Scheme, mURL.Host+mURL.Path)65			}66			if j.server.NetPoolConnections > 0 {67				outsrv.SetMaxConnections(j.server.NetPoolConnections)68			}69			// populate it70			err = outsrv.InitPool()71			if err != nil {72				j.server.log.Warning("Poll init error %s", err)73				return 174			}75			j.server.Outpool[j.outserver] = outsrv76			return 077		}78	}79	// keep retrying80	retcode := make_pool()81	if retcode == 1 {82		time.Sleep(time.Second)83		return fmt.Errorf("Pool %s failed to intialize, check the outgoing servers for 'aliveness'", j.outserver)84	} else if retcode == 2 {85		return fmt.Errorf("Pool %s failed to intialize, Hard failure", j.outserver)86	}87	netconn, err := outsrv.Open()88	defer outsrv.Close(netconn)89	if err != nil {90		stats.StatsdClient.Incr("failed.bad-connection", 1)91		j.server.FailSendCount.Up(1)92		return fmt.Errorf("Error sending to backend %s", err)93	}94	if netconn.Conn() != nil {95		// Conn.Write will raise a timeout error after 1 seconds96		netconn.SetWriteDeadline(time.Now().Add(j.server.WriteTimeout))97		//var wrote int98		toSend := append(j.param, repr.NEWLINE_SEPARATOR_BYTE)99		by, err := netconn.Write(toSend)100		//log.Printf("SEND %s %s", wrote, err)101		if err != nil {102			stats.StatsdClient.Incr("failed.connection-timeout", 1)103			j.server.FailSendCount.Up(1)104			outsrv.ResetConn(netconn)105			return fmt.Errorf("Error sending (writing) to backend: %s", err)106		} else {107			j.server.BytesWrittenCount.Up(uint64(by))108			j.server.SuccessSendCount.Up(1)109			stats.StatsdClient.Incr("success.send", 1)110			stats.StatsdClient.Incr("success.sent-bytes", int64(len(toSend)))111		}112	} else {113		stats.StatsdClient.Incr("failed.aborted-connection", 1)114		j.server.FailSendCount.Up(1)115		return fmt.Errorf("Error sending (writing connection gone) to backend: %s", j.outserver)116	}117	return nil118}119/** SingleWriter a writer that does not use a buffered pool for sending to sockets **/120type SingleWriter struct{}121func (p *SingleWriter) Write(j *OutputMessage) error {122	//this is for using a simple tcp connection per stat we send out123	//one can quickly run out of sockets if this is used under high load124	if j == nil {125		return nil126	}127	defer stats.StatsdNanoTimeFunc("worker.process-time-ns", time.Now())128	m_url, err := url.Parse(j.outserver)129	if err != nil {130		stats.StatsdClient.Incr("failed.bad-url", 1)131		j.server.FailSendCount.Up(1)132		return fmt.Errorf("Error sending to backend Invalid URL %s", err)133	}134	conn, err := net.DialTimeout(m_url.Scheme, m_url.Host+m_url.Path, 5*time.Second)135	if conn != nil {136		conn.SetWriteDeadline(time.Now().Add(j.server.WriteTimeout))137		//send it and close it138		to_send := append(j.param, repr.NEWLINE_SEPARATOR_BYTE)139		_, err = conn.Write(to_send)140		conn.Close()141		conn = nil142		if err != nil {143			stats.StatsdClient.Incr("failed.bad-connection", 1)144			j.server.FailSendCount.Up(1)145			return fmt.Errorf("Error sending (writing) to backend: %s", err)146		}147		stats.StatsdClient.Incr("success.sent", 1)148		stats.StatsdClient.Incr("success.sent-bytes", int64(len(to_send)))149		j.server.SuccessSendCount.Up(1)150	} else {151		j.server.FailSendCount.Up(1)152		return fmt.Errorf("Error sending (connection) to backend: %s", err)153	}154	return nil155}156/***** OutputDispatchJob Dispatcher Job work queue ****/157type OutputDispatchJob struct {158	Writer  OutMessageWriter159	Message *OutputMessage160	Retry   int161}162func (o *OutputDispatchJob) IncRetry() int {163	o.Retry++164	return o.Retry165}166func (o *OutputDispatchJob) OnRetry() int {167	return o.Retry168}169func (o *OutputDispatchJob) DoWork() error {170	err := o.Writer.Write(o.Message)171	if err != nil {172		o.Message.server.log.Error("%s", err)173		if o.OnRetry() < 2 {174			o.Message.server.log.Warning("Retrying message: %s retry #%d", o.Message.param, o.OnRetry())175		}176	}177	return err178}179func NewOutputDispatcher(workers int) *dispatch.Dispatch {180	writeQueue := make(chan dispatch.IJob, workers*10) // a little buffer181	dispatchQueue := make(chan chan dispatch.IJob, workers)182	writeDispatcher := dispatch.NewDispatch(workers, dispatchQueue, writeQueue)183	writeDispatcher.SetRetries(2)184	return writeDispatcher185}...

relay.go

Source:relay.go

...96			}97			if !graphiteDispatchStarted && strings.Contains(metric.Namespace.String(), "collectd") {98				r.graphiteServer.Start()99				graphiteDispatchStarted = true100				go dispatchMetrics(ctx, r.graphiteServer.Metrics(ctx), metrics_out)101			}102			if !statsdDispatchStarted && strings.Contains(metric.Namespace.String(), "statsd") {103				r.statsdServer.Start()104				statsdDispatchStarted = true105				go dispatchMetrics(ctx, r.statsdServer.Metrics(ctx), metrics_out)106			}107		}108	}109	return nil110}111func dispatchMetrics(ctx context.Context, in chan *plugin.Metric, out chan []plugin.Metric) {112	for {113		select {114		case metric := <-in:115			log.WithFields(116				log.Fields{117					"metric": metric.Namespace.String(),118					"data":   metric.Data,119				},120			).Debug("dispatching metrics")121			out <- []plugin.Metric{*metric}122		case <-ctx.Done():123			return124		}125	}126}127/*128	GetMetricTypes() returns the metric types for testing129	GetMetricTypes() will be called when your plugin is loaded in order to populate the metric catalog(where snaps stores all130	available metrics).131	Config info is passed in. This config information would come from global config snap settings.132	The metrics returned will be advertised to users who list all the metrics and will become targetable by tasks.133*/134func (r *relay) GetMetricTypes(cfg plugin.Config) ([]plugin.Metric, error) {...

metrics.go

Source:metrics.go

1package interceptor2import (3	"context"4	"fmt"5	"time"6	"github.com/gojek/fiber"7)8// StatsdClient is an interface for a stats listener9type StatsdClient interface {10	Increment(string)11	Unique(string, string)12	Timing(string, interface{})13}14// MetricsKey is an alias for a string15type MetricsKey string16// CtxDispatchStartTimeKey is used to record the start time of a request17var (18	CtxDispatchStartTimeKey MetricsKey = "CTX_DISPATCH_START_TIME"19)20// NewMetricsInterceptor creates a new MetricsInterceptor with the given client21func NewMetricsInterceptor(client StatsdClient) fiber.Interceptor {22	return &MetricsInterceptor{23		statsd: client,24	}25}26// MetricsInterceptor is an interceptor to log metrics27type MetricsInterceptor struct {28	fiber.NoopAfterDispatchInterceptor29	statsd StatsdClient30}31func (i *MetricsInterceptor) operationName(ctx context.Context, req fiber.Request, suffix string) string {32	componentID := ctx.Value(fiber.CtxComponentIDKey)33	return fmt.Sprintf("fiber.%s.%s", componentID, suffix)34}35// BeforeDispatch records count of the requests through a component and the request start time36func (i *MetricsInterceptor) BeforeDispatch(ctx context.Context, req fiber.Request) context.Context {37	i.statsd.Increment(i.operationName(ctx, req, "count"))38	ctx = context.WithValue(ctx, CtxDispatchStartTimeKey, time.Now())39	return ctx40}41// AfterCompletion computes and submits the request completion time42func (i *MetricsInterceptor) AfterCompletion(ctx context.Context, req fiber.Request, queue fiber.ResponseQueue) {43	if startTime, ok := ctx.Value(CtxDispatchStartTimeKey).(time.Time); ok {44		i.statsd.Timing(i.operationName(ctx, req, "timing"), int(time.Since(startTime)/time.Millisecond))45	}46}...

dispatch

Using AI Code Generation

1import (2func main() {3    statsd_client := statsd.NewStatsdClient("localhost:8125", "example")4    statsd_client.CreateSocket()5    defer statsd_client.Close()6    for i := 0; i < 10; i++ {7        statsd_client.Incr("test.counter", 1)8        statsd_client.Timing("test.time", 50)9        statsd_client.PrecisionTiming("test.precisiontime", 50)10        statsd_client.Gauge("test.gauge", 10)11        statsd_client.GaugeDelta("test.gaugedelta", 10)12        statsd_client.GaugeDelta("test.gaugedelta", -5)13        statsd_client.GaugePercent("test.gaugepercent", 50)14        statsd_client.GaugePercent("test.gaugepercent", 100)15        statsd_client.GaugePercent("test.gaugepercent", 0)16        statsd_client.GaugeDeltaPercent("test.gaugedeltapercent", 50)17        statsd_client.GaugeDeltaPercent("test.gaugedeltapercent", 100)18        statsd_client.GaugeDeltaPercent("test.gaugedeltapercent", -100)19        statsd_client.GaugeDeltaPercent("test.gaugedeltapercent", 0)20        statsd_client.FGauge("test.fgauge", 10.123)21        statsd_client.FGaugeDelta("test.fgaugedelta", 10.123)22        statsd_client.FGaugeDelta("test.fgaugedelta", -5.123)23        statsd_client.FGaugePercent("test.fgaugepercent", 50.123)24        statsd_client.FGaugePercent("test.fgaugepercent", 100.123)25        statsd_client.FGaugePercent("test.fgaugepercent", 0.123)26        statsd_client.FGaugeDeltaPercent("test.fgaugedeltapercent", 50.123)27        statsd_client.FGaugeDeltaPercent("test.fgaugedeltapercent", 100.123)28        statsd_client.FGaugeDeltaPercent("test.fgaugedeltapercent", -100.123)29        statsd_client.FGaugeDeltaPercent("test.fgaugedeltapercent", 0

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.