@@ -26,142 +26,126 @@ import (
2626 "go.opentelemetry.io/collector/pdata/plog"
2727 "go.opentelemetry.io/collector/pdata/pmetric"
2828 "go.opentelemetry.io/collector/pdata/ptrace"
29+ "go.uber.org/multierr"
2930)
3031
3132type tracesExporter struct {
3233 logger common.Logger
33- cfg * Config
3434 writer * influxHTTPWriter
3535 converter * otel2influx.OtelTracesToLineProtocol
36- settings component.TelemetrySettings
3736}
3837
39- func newTracesExporter (config * Config , params exporter.CreateSettings ) * tracesExporter {
40- logger := newZapInfluxLogger (params .Logger )
41- converter := otel2influx .NewOtelTracesToLineProtocol (logger )
38+ func newTracesExporter (config * Config , settings exporter.CreateSettings ) (* tracesExporter , error ) {
39+ logger := newZapInfluxLogger (settings .Logger )
40+
41+ writer , err := newInfluxHTTPWriter (logger , config , settings .TelemetrySettings )
42+ if err != nil {
43+ return nil , err
44+ }
45+
46+ converter , err := otel2influx .NewOtelTracesToLineProtocol (logger , writer )
47+ if err != nil {
48+ return nil , err
49+ }
4250
4351 return & tracesExporter {
4452 logger : logger ,
45- cfg : config ,
53+ writer : writer ,
4654 converter : converter ,
47- settings : params .TelemetrySettings ,
48- }
55+ }, nil
4956}
5057
5158func (e * tracesExporter ) pushTraces (ctx context.Context , td ptrace.Traces ) error {
52- batch := e .writer .newBatch ()
53-
54- err := e .converter .WriteTraces (ctx , td , batch )
59+ err := e .converter .WriteTraces (ctx , td )
5560 if err != nil {
5661 return consumererror .NewPermanent (err )
5762 }
58- return batch . flushAndClose ( ctx )
63+ return nil
5964}
6065
61- // start starts the traces exporter
62- func (e * tracesExporter ) start (_ context.Context , host component.Host ) (err error ) {
63-
64- writer , err := newInfluxHTTPWriter (e .logger , e .cfg , host , e .settings )
65- if err != nil {
66- return err
67- }
68- e .writer = writer
66+ func (e * tracesExporter ) Start (ctx context.Context , host component.Host ) error {
67+ e .logger .Debug ("starting traces exporter" )
68+ return multierr .Combine (
69+ e .writer .Start (ctx , host ),
70+ e .converter .Start (ctx , host ))
71+ }
6972
70- return nil
73+ func (e * tracesExporter ) Shutdown (ctx context.Context ) error {
74+ return e .converter .Shutdown (ctx )
7175}
7276
7377type metricsExporter struct {
7478 logger common.Logger
75- cfg * Config
7679 writer * influxHTTPWriter
7780 converter * otel2influx.OtelMetricsToLineProtocol
78- settings component.TelemetrySettings
79- }
80-
81- var metricsSchemata = map [string ]common.MetricsSchema {
82- "telegraf-prometheus-v1" : common .MetricsSchemaTelegrafPrometheusV1 ,
83- "telegraf-prometheus-v2" : common .MetricsSchemaTelegrafPrometheusV2 ,
8481}
8582
8683func newMetricsExporter (config * Config , params exporter.CreateSettings ) (* metricsExporter , error ) {
8784 logger := newZapInfluxLogger (params .Logger )
88- schema , found := metricsSchemata [config .MetricsSchema ]
85+ schema , found := common . MetricsSchemata [config .MetricsSchema ]
8986 if ! found {
9087 return nil , fmt .Errorf ("schema '%s' not recognized" , config .MetricsSchema )
9188 }
9289
93- converter , err := otel2influx .NewOtelMetricsToLineProtocol (logger , schema )
90+ writer , err := newInfluxHTTPWriter (logger , config , params .TelemetrySettings )
91+ if err != nil {
92+ return nil , err
93+ }
94+
95+ converter , err := otel2influx .NewOtelMetricsToLineProtocol (logger , writer , schema )
9496 if err != nil {
9597 return nil , err
9698 }
9799
98100 return & metricsExporter {
99101 logger : logger ,
100- cfg : config ,
102+ writer : writer ,
101103 converter : converter ,
102- settings : params .TelemetrySettings ,
103104 }, nil
104105}
105106
106107func (e * metricsExporter ) pushMetrics (ctx context.Context , md pmetric.Metrics ) error {
107- batch := e .writer .newBatch ()
108-
109- err := e .converter .WriteMetrics (ctx , md , batch )
108+ err := e .converter .WriteMetrics (ctx , md )
110109 if err != nil {
111110 return consumererror .NewPermanent (err )
112111 }
113- return batch . flushAndClose ( ctx )
112+ return nil
114113}
115114
116- // start starts the metrics exporter
117- func (e * metricsExporter ) start (_ context.Context , host component.Host ) (err error ) {
118-
119- writer , err := newInfluxHTTPWriter (e .logger , e .cfg , host , e .settings )
120- if err != nil {
121- return err
122- }
123- e .writer = writer
124-
125- return nil
115+ func (e * metricsExporter ) Start (ctx context.Context , host component.Host ) error {
116+ return e .writer .Start (ctx , host )
126117}
127118
128119type logsExporter struct {
129120 logger common.Logger
130- cfg * Config
131121 writer * influxHTTPWriter
132122 converter * otel2influx.OtelLogsToLineProtocol
133- settings component.TelemetrySettings
134123}
135124
136- func newLogsExporter (config * Config , params exporter.CreateSettings ) * logsExporter {
125+ func newLogsExporter (config * Config , params exporter.CreateSettings ) ( * logsExporter , error ) {
137126 logger := newZapInfluxLogger (params .Logger )
138- converter := otel2influx .NewOtelLogsToLineProtocol (logger )
127+
128+ writer , err := newInfluxHTTPWriter (logger , config , params .TelemetrySettings )
129+ if err != nil {
130+ return nil , err
131+ }
132+ converter := otel2influx .NewOtelLogsToLineProtocol (logger , writer )
139133
140134 return & logsExporter {
141135 logger : logger ,
136+ writer : writer ,
142137 converter : converter ,
143- cfg : config ,
144- settings : params .TelemetrySettings ,
145- }
138+ }, nil
146139}
147140
148141func (e * logsExporter ) pushLogs (ctx context.Context , ld plog.Logs ) error {
149- batch := e .writer .newBatch ()
150-
151- err := e .converter .WriteLogs (ctx , ld , batch )
142+ err := e .converter .WriteLogs (ctx , ld )
152143 if err != nil {
153144 return consumererror .NewPermanent (err )
154145 }
155- return batch . flushAndClose ( ctx )
146+ return nil
156147}
157148
158- // start starts the logs exporter
159- func (e * logsExporter ) start (_ context.Context , host component.Host ) (err error ) {
160- writer , err := newInfluxHTTPWriter (e .logger , e .cfg , host , e .settings )
161- if err != nil {
162- return err
163- }
164- e .writer = writer
165-
166- return nil
149+ func (e * logsExporter ) Start (ctx context.Context , host component.Host ) error {
150+ return e .writer .Start (ctx , host )
167151}
0 commit comments