Skip to content

Commit

Permalink
Revert "[exporter/signalfx] Add option to send histograms in OTLP for…
Browse files Browse the repository at this point in the history
…mat (open-telemetry#31197)"

This reverts commit 71c5c3c.
  • Loading branch information
vasireddy99 committed May 7, 2024
1 parent fd82441 commit 0f75d32
Show file tree
Hide file tree
Showing 16 changed files with 81 additions and 1,305 deletions.
3 changes: 1 addition & 2 deletions exporter/signalfxexporter/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -145,8 +145,7 @@ will be replaced with a `_`.
api_tls:
ca_file: "/etc/opt/certs/ca.pem"
```
- `drop_histogram_buckets`: (default = `false`) if set to true, histogram buckets will not be translated into datapoints with `_bucket` suffix but will be dropped instead, only datapoints with `_sum`, `_count`, `_min` (optional) and `_max` (optional) suffixes will be sent. Please note that this option does not apply to histograms sent in OTLP format with `send_otlp_histograms` enabled.
- `send_otlp_histograms`: (default: `false`) if set to true, any histogram metrics receiver by the exporter will be sent to Splunk Observability backend in OTLP format without conversion to SignalFx format. This can only be enabled if the Splunk Observability environment (realm) has the new Histograms feature rolled out. Please note that histograms sent in OTLP format do not apply to the exporter configurations `include_metrics` and `exclude_metrics`.
- `drop_histogram_buckets`: (default = `false`) if set to true, histogram buckets will not be translated into datapoints with `_bucket` suffix but will be dropped instead, only datapoints with `_sum`, `_count`, `_min` (optional) and `_max` (optional) suffixes will be sent.
In addition, this exporter offers queued retry which is enabled by default.
Information about queued retry configuration parameters can be found
[here](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md).
Expand Down
6 changes: 1 addition & 5 deletions exporter/signalfxexporter/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ type Config struct {

// ExcludeMetrics defines dpfilter.MetricFilters that will determine metrics to be
// excluded from sending to SignalFx backend. If translations enabled with
// TranslationRules options, the exclusion will be applied on translated metrics.
// TranslationRules options, the exclusion will be applie on translated metrics.
ExcludeMetrics []dpfilters.MetricFilter `mapstructure:"exclude_metrics"`

// IncludeMetrics defines dpfilter.MetricFilters to override exclusion any of metric.
Expand All @@ -134,10 +134,6 @@ type Config struct {
// Whether to drop histogram bucket metrics dispatched to Splunk Observability.
// Default value is set to false.
DropHistogramBuckets bool `mapstructure:"drop_histogram_buckets"`

// Whether to send histogram metrics in OTLP format to Splunk Observability.
// Default value is set to false.
SendOTLPHistograms bool `mapstructure:"send_otlp_histograms"`
}

type DimensionClientConfig struct {
Expand Down
2 changes: 0 additions & 2 deletions exporter/signalfxexporter/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,6 @@ func TestLoadConfig(t *testing.T) {
},
},
NonAlphanumericDimensionChars: "_-.",
SendOTLPHistograms: false,
},
},
{
Expand Down Expand Up @@ -264,7 +263,6 @@ func TestLoadConfig(t *testing.T) {
},
},
NonAlphanumericDimensionChars: "_-.",
SendOTLPHistograms: true,
},
},
}
Expand Down
144 changes: 19 additions & 125 deletions exporter/signalfxexporter/dpclient.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,20 +17,12 @@ import (
sfxpb "github.com/signalfx/com_signalfx_metrics_protobuf/model"
"go.opentelemetry.io/collector/consumer/consumererror"
"go.opentelemetry.io/collector/pdata/pmetric"
"go.opentelemetry.io/collector/pdata/pmetric/pmetricotlp"
"go.uber.org/zap"

"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/signalfxexporter/internal/translation"
"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/signalfxexporter/internal/utils"
"github.com/open-telemetry/opentelemetry-collector-contrib/internal/splunk"
)

const (
contentEncodingHeader = "Content-Encoding"
contentTypeHeader = "Content-Type"
otlpProtobufContentType = "application/x-protobuf;format=otlp"
)

type sfxClientBase struct {
ingestURL *url.URL
headers map[string]string
Expand Down Expand Up @@ -66,7 +58,6 @@ type sfxDPClient struct {
logger *zap.Logger
accessTokenPassthrough bool
converter *translation.MetricsConverter
sendOTLPHistograms bool
}

func (s *sfxDPClient) pushMetricsData(
Expand All @@ -90,55 +81,48 @@ func (s *sfxDPClient) pushMetricsData(
// All metrics in the pmetric.Metrics will have the same access token because of the BatchPerResourceMetrics.
metricToken := s.retrieveAccessToken(rms.At(0))

// export SFx format
sfxDataPoints := s.converter.MetricsToSignalFxV2(md)
if len(sfxDataPoints) > 0 {
droppedCount, err := s.pushMetricsDataForToken(ctx, sfxDataPoints, metricToken)
if err != nil {
return droppedCount, err
if s.logDataPoints {
for _, dp := range sfxDataPoints {
s.logger.Debug("Dispatching SFx datapoint", zap.Stringer("dp", dp))
}
}
return s.pushMetricsDataForToken(ctx, sfxDataPoints, metricToken)
}

// export any histograms in otlp if sendOTLPHistograms is true
if s.sendOTLPHistograms {
histogramData, metricCount := utils.GetHistograms(md)
if metricCount > 0 {
droppedCount, err := s.pushOTLPMetricsDataForToken(ctx, histogramData, metricToken)
if err != nil {
return droppedCount, err
}
}
func (s *sfxDPClient) pushMetricsDataForToken(ctx context.Context, sfxDataPoints []*sfxpb.DataPoint, accessToken string) (int, error) {
body, compressed, err := s.encodeBody(sfxDataPoints)
if err != nil {
return len(sfxDataPoints), consumererror.NewPermanent(err)
}

return 0, nil

}

func (s *sfxDPClient) postData(ctx context.Context, body io.Reader, headers map[string]string) error {
datapointURL := *s.ingestURL
if !strings.HasSuffix(datapointURL.Path, "v2/datapoint") {
datapointURL.Path = path.Join(datapointURL.Path, "v2/datapoint")
}
req, err := http.NewRequestWithContext(ctx, "POST", datapointURL.String(), body)
if err != nil {
return consumererror.NewPermanent(err)
return len(sfxDataPoints), consumererror.NewPermanent(err)
}

// Set the headers configured in sfxDPClient
for k, v := range s.headers {
req.Header.Set(k, v)
}

// Set any extra headers passed by the caller
for k, v := range headers {
req.Header.Set(k, v)
// Override access token in headers map if it's non empty.
if accessToken != "" {
req.Header.Set(splunk.SFxAccessTokenHeader, accessToken)
}

if compressed {
req.Header.Set("Content-Encoding", "gzip")
}

// TODO: Mark errors as partial errors wherever applicable when, partial
// error for metrics is available.
resp, err := s.client.Do(req)
if err != nil {
return err
return len(sfxDataPoints), err
}

defer func() {
Expand All @@ -148,39 +132,7 @@ func (s *sfxDPClient) postData(ctx context.Context, body io.Reader, headers map[

err = splunk.HandleHTTPCode(resp)
if err != nil {
return err
}
return nil
}

func (s *sfxDPClient) pushMetricsDataForToken(ctx context.Context, sfxDataPoints []*sfxpb.DataPoint, accessToken string) (int, error) {

if s.logDataPoints {
for _, dp := range sfxDataPoints {
s.logger.Debug("Dispatching SFx datapoint", zap.Stringer("dp", dp))
}
}

body, compressed, err := s.encodeBody(sfxDataPoints)
dataPointCount := len(sfxDataPoints)
if err != nil {
return dataPointCount, consumererror.NewPermanent(err)
}

headers := make(map[string]string)

// Override access token in headers map if it's non empty.
if accessToken != "" {
headers[splunk.SFxAccessTokenHeader] = accessToken
}

if compressed {
headers[contentEncodingHeader] = "gzip"
}

err = s.postData(ctx, body, headers)
if err != nil {
return dataPointCount, err
return len(sfxDataPoints), err
}
return 0, nil
}
Expand Down Expand Up @@ -208,61 +160,3 @@ func (s *sfxDPClient) retrieveAccessToken(md pmetric.ResourceMetrics) string {
}
return ""
}

func (s *sfxDPClient) pushOTLPMetricsDataForToken(ctx context.Context, mh pmetric.Metrics, accessToken string) (int, error) {

dataPointCount := mh.DataPointCount()
if s.logDataPoints {
s.logger.Debug("Count of metrics to send in OTLP format",
zap.Int("resource metrics", mh.ResourceMetrics().Len()),
zap.Int("metrics", mh.MetricCount()),
zap.Int("data points", dataPointCount))
buf, err := metricsMarshaler.MarshalMetrics(mh)
if err != nil {
s.logger.Error("Failed to marshal metrics for logging otlp histograms", zap.Error(err))
} else {
s.logger.Debug("Dispatching OTLP metrics", zap.String("pmetrics", string(buf)))
}
}

body, compressed, err := s.encodeOTLPBody(mh)
if err != nil {
return dataPointCount, consumererror.NewPermanent(err)
}

headers := make(map[string]string)

// Set otlp content-type header
headers[contentTypeHeader] = otlpProtobufContentType

// Override access token in headers map if it's non-empty.
if accessToken != "" {
headers[splunk.SFxAccessTokenHeader] = accessToken
}

if compressed {
headers[contentEncodingHeader] = "gzip"
}

s.logger.Debug("Sending metrics in OTLP format")

err = s.postData(ctx, body, headers)

if err != nil {
return dataPointCount, consumererror.NewMetrics(err, mh)
}

return 0, nil
}

func (s *sfxDPClient) encodeOTLPBody(md pmetric.Metrics) (bodyReader io.Reader, compressed bool, err error) {

tr := pmetricotlp.NewExportRequestFromMetrics(md)

body, err := tr.MarshalProto()

if err != nil {
return nil, false, err
}
return s.getReader(body)
}
2 changes: 0 additions & 2 deletions exporter/signalfxexporter/exporter.go
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,6 @@ func newSignalFxExporter(
config.IncludeMetrics,
config.NonAlphanumericDimensionChars,
config.DropHistogramBuckets,
!config.SendOTLPHistograms, // if SendOTLPHistograms is true, do not process histograms when converting to SFx
)
if err != nil {
return nil, fmt.Errorf("failed to create metric converter: %w", err)
Expand Down Expand Up @@ -122,7 +121,6 @@ func (se *signalfxExporter) start(ctx context.Context, host component.Host) (err
logger: se.logger,
accessTokenPassthrough: se.config.AccessTokenPassthrough,
converter: se.converter,
sendOTLPHistograms: se.config.SendOTLPHistograms,
}

apiTLSCfg, err := se.config.APITLSSettings.LoadTLSConfig()
Expand Down
Loading

0 comments on commit 0f75d32

Please sign in to comment.