Need help in creating an exporter to directly publish metrics to Victoria Metrics

[ayush-san]

Hi team,

I have been trying to setup a collector that will directly publish metrics our Victoria cluster using the open-telemetry metrics format. Victoria has provided an example code but it was written using old version of otel metrics library - https://docs.victoriametrics.com/guides/getting-started-with-opentelemetry-app.go.example

Flink jobs will publish data to a kafka topic and my application will publish it to Victoria. I have tried to write my own using the new version and it is working

func NewExporterClient(ctx context.Context, cfg *Config) (*Client, error) {
	options := []otlpmetrichttp.Option{
		otlpmetrichttp.WithEndpoint(cfg.Endpoint),
		otlpmetrichttp.WithURLPath(cfg.Url),
	}

	if !cfg.IsSecure {
		options = append(options, otlpmetrichttp.WithInsecure())
	}

	switch cfg.Compression {
	case "gzip":
		options = append(options, otlpmetrichttp.WithCompression(otlpmetrichttp.GzipCompression))
	}

	metricExporter, err := otlpmetrichttp.New(ctx, options...)
	if err != nil {
		return nil, err
	}

	// Create resource.
	res, err := resource.Merge(resource.Default(),
		resource.NewSchemaless(
			attribute.Key("job").String(cfg.JobName),
			attribute.Key("instance").String(cfg.InstanceName),
		),
	)
	if err != nil {
		return nil, err
	}

	periodicReader := metric.NewPeriodicReader(metricExporter,
		metric.WithInterval(time.Duration(cfg.PushInterval)*time.Second),
		metric.WithTimeout(time.Duration(cfg.PushTimeout)*time.Second),
	)

	provider := metric.NewMeterProvider(metric.WithReader(periodicReader), metric.WithResource(res))

	return &Client{
		Meter:  provider.Meter(cfg.JobName),
		Config: cfg,
	}, nil
}

func (client *Client) publishGauge(ctx context.Context, metricName string, payload *Metric) error {
	attributes := make([]attribute.KeyValue, 0, len(payload.Labels))
	for k, v := range payload.Labels {
		attributes = append(attributes, attribute.Key(k).String(v))
	}
	gauge, err := client.Meter.Float64Gauge(metricName, api.WithDescription(payload.Help))
	if err != nil {
		log.WithError(err).WithFields(map[string]interface{}{
			"event_name":  "publish_gauge",
			"namespace":   payload.Namespace,
			"subsystem":   payload.Subsystem,
			"metric_name": payload.MetricName,
		}).Error("Error pushing gauge metric")
		return err
	}
	gauge.Record(ctx, payload.MetricValue, api.WithAttributes(attributes...))
	return nil
}

But I have some doubts regarding the above flow

• Since this will periodically push metrics to Victoria, how to stats of the metrics it published in every interval
• Success ratio/count of the metrics published
• How to publish timestamp of the metrics

While testing my code, I overwhelmed my victoria cluster and it went down but I didn't know how to propagate it to my consumer so that it also stop consuming new events

2024/06/01 07:07:29 failed to upload metrics: context deadline exceeded: retry-able request failure
2024/06/01 07:07:49 failed to upload metrics: context deadline exceeded: retry-able request failure