|
| 1 | +// Copyright The OpenTelemetry Authors |
| 2 | +// SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +package otelarrowexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/otelarrowexporter" |
| 5 | + |
| 6 | +import ( |
| 7 | + "context" |
| 8 | + "errors" |
| 9 | + "fmt" |
| 10 | + "runtime" |
| 11 | + "sort" |
| 12 | + "strings" |
| 13 | + "sync" |
| 14 | + |
| 15 | + arrowPkg "github.com/apache/arrow/go/v16/arrow" |
| 16 | + "go.opentelemetry.io/collector/client" |
| 17 | + "go.opentelemetry.io/collector/component" |
| 18 | + "go.opentelemetry.io/collector/consumer/consumererror" |
| 19 | + "go.opentelemetry.io/collector/exporter" |
| 20 | + "go.opentelemetry.io/collector/pdata/plog" |
| 21 | + "go.opentelemetry.io/collector/pdata/pmetric" |
| 22 | + "go.opentelemetry.io/collector/pdata/ptrace" |
| 23 | + "go.opentelemetry.io/otel/attribute" |
| 24 | + "go.uber.org/multierr" |
| 25 | + "google.golang.org/grpc/metadata" |
| 26 | + |
| 27 | + "github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/compression/zstd" |
| 28 | + "github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/netstats" |
| 29 | +) |
| 30 | + |
| 31 | +var ( |
| 32 | + // errTooManyExporters is returned when the MetadataCardinalityLimit has been reached. |
| 33 | + errTooManyExporters = consumererror.NewPermanent(errors.New("too many exporter metadata-value combinations")) |
| 34 | +) |
| 35 | + |
| 36 | +type metadataExporter struct { |
| 37 | + config *Config |
| 38 | + settings exporter.Settings |
| 39 | + scf streamClientFactory |
| 40 | + host component.Host |
| 41 | + |
| 42 | + metadataKeys []string |
| 43 | + exporters sync.Map |
| 44 | + netReporter *netstats.NetworkReporter |
| 45 | + |
| 46 | + userAgent string |
| 47 | + |
| 48 | + // Guards the size and the storing logic to ensure no more than limit items are stored. |
| 49 | + // If we are willing to allow "some" extra items than the limit this can be removed and size can be made atomic. |
| 50 | + lock sync.Mutex |
| 51 | + size int |
| 52 | +} |
| 53 | + |
| 54 | +var _ exp = (*metadataExporter)(nil) |
| 55 | + |
| 56 | +func newMetadataExporter(cfg component.Config, set exporter.Settings, streamClientFactory streamClientFactory) (exp, error) { |
| 57 | + oCfg := cfg.(*Config) |
| 58 | + netReporter, err := netstats.NewExporterNetworkReporter(set) |
| 59 | + if err != nil { |
| 60 | + return nil, err |
| 61 | + } |
| 62 | + userAgent := fmt.Sprintf("%s/%s (%s/%s)", |
| 63 | + set.BuildInfo.Description, set.BuildInfo.Version, runtime.GOOS, runtime.GOARCH) |
| 64 | + |
| 65 | + if !oCfg.Arrow.Disabled { |
| 66 | + // Ignoring an error because Validate() was called. |
| 67 | + _ = zstd.SetEncoderConfig(oCfg.Arrow.Zstd) |
| 68 | + |
| 69 | + userAgent += fmt.Sprintf(" ApacheArrow/%s (NumStreams/%d)", arrowPkg.PkgVersion, oCfg.Arrow.NumStreams) |
| 70 | + } |
| 71 | + // use lower-case, to be consistent with http/2 headers. |
| 72 | + mks := make([]string, len(oCfg.MetadataKeys)) |
| 73 | + for i, k := range oCfg.MetadataKeys { |
| 74 | + mks[i] = strings.ToLower(k) |
| 75 | + } |
| 76 | + sort.Strings(mks) |
| 77 | + if len(mks) == 0 { |
| 78 | + return newExporter(cfg, set, streamClientFactory, userAgent, netReporter) |
| 79 | + } |
| 80 | + return &metadataExporter{ |
| 81 | + config: oCfg, |
| 82 | + settings: set, |
| 83 | + scf: streamClientFactory, |
| 84 | + metadataKeys: mks, |
| 85 | + userAgent: userAgent, |
| 86 | + netReporter: netReporter, |
| 87 | + }, nil |
| 88 | +} |
| 89 | + |
| 90 | +func (e *metadataExporter) getSettings() exporter.Settings { |
| 91 | + return e.settings |
| 92 | +} |
| 93 | + |
| 94 | +func (e *metadataExporter) getConfig() component.Config { |
| 95 | + return e.config |
| 96 | +} |
| 97 | + |
| 98 | +func (e *metadataExporter) start(_ context.Context, host component.Host) (err error) { |
| 99 | + e.host = host |
| 100 | + return nil |
| 101 | +} |
| 102 | + |
| 103 | +func (e *metadataExporter) shutdown(ctx context.Context) error { |
| 104 | + var err error |
| 105 | + e.exporters.Range(func(_ any, value any) bool { |
| 106 | + be := value.(exp) |
| 107 | + err = multierr.Append(err, be.shutdown(ctx)) |
| 108 | + return true |
| 109 | + }) |
| 110 | + return err |
| 111 | +} |
| 112 | + |
| 113 | +func (e *metadataExporter) pushTraces(ctx context.Context, td ptrace.Traces) error { |
| 114 | + s, mdata := e.getAttrSet(ctx, e.metadataKeys) |
| 115 | + |
| 116 | + be, err := e.getOrCreateExporter(ctx, s, mdata) |
| 117 | + if err != nil { |
| 118 | + return err |
| 119 | + } |
| 120 | + return be.pushTraces(ctx, td) |
| 121 | +} |
| 122 | + |
| 123 | +func (e *metadataExporter) pushMetrics(ctx context.Context, md pmetric.Metrics) error { |
| 124 | + s, mdata := e.getAttrSet(ctx, e.metadataKeys) |
| 125 | + |
| 126 | + be, err := e.getOrCreateExporter(ctx, s, mdata) |
| 127 | + if err != nil { |
| 128 | + return err |
| 129 | + } |
| 130 | + |
| 131 | + return be.pushMetrics(ctx, md) |
| 132 | +} |
| 133 | + |
| 134 | +func (e *metadataExporter) pushLogs(ctx context.Context, ld plog.Logs) error { |
| 135 | + s, mdata := e.getAttrSet(ctx, e.metadataKeys) |
| 136 | + |
| 137 | + be, err := e.getOrCreateExporter(ctx, s, mdata) |
| 138 | + if err != nil { |
| 139 | + return err |
| 140 | + } |
| 141 | + |
| 142 | + return be.pushLogs(ctx, ld) |
| 143 | +} |
| 144 | + |
| 145 | +func (e *metadataExporter) getOrCreateExporter(ctx context.Context, s attribute.Set, md metadata.MD) (exp, error) { |
| 146 | + e.lock.Lock() |
| 147 | + defer e.lock.Unlock() |
| 148 | + |
| 149 | + if e.config.MetadataCardinalityLimit != 0 && e.size >= int(e.config.MetadataCardinalityLimit) { |
| 150 | + return nil, errTooManyExporters |
| 151 | + } |
| 152 | + |
| 153 | + v, ok := e.exporters.Load(s) |
| 154 | + if ok { |
| 155 | + return v.(exp), nil |
| 156 | + } |
| 157 | + |
| 158 | + newExp, err := newExporter(e.config, e.settings, e.scf, e.userAgent, e.netReporter) |
| 159 | + if err != nil { |
| 160 | + return nil, fmt.Errorf("failed to create exporter: %w", err) |
| 161 | + } |
| 162 | + |
| 163 | + var loaded bool |
| 164 | + v, loaded = e.exporters.LoadOrStore(s, newExp) |
| 165 | + if !loaded { |
| 166 | + // set metadata keys for base exporter to add them to the outgoing context. |
| 167 | + newExp.(*baseExporter).setMetadata(md) |
| 168 | + |
| 169 | + // Start the goroutine only if we added the object to the map, otherwise is already started. |
| 170 | + err = newExp.start(ctx, e.host) |
| 171 | + if err != nil { |
| 172 | + e.exporters.Delete(s) |
| 173 | + return nil, fmt.Errorf("failed to start exporter: %w", err) |
| 174 | + } |
| 175 | + |
| 176 | + e.size++ |
| 177 | + } |
| 178 | + |
| 179 | + return v.(exp), nil |
| 180 | +} |
| 181 | + |
| 182 | +// getAttrSet is code taken from the core collector's batchprocessor multibatch logic. |
| 183 | +// https://github.com/open-telemetry/opentelemetry-collector/blob/v0.107.0/processor/batchprocessor/batch_processor.go#L298 |
| 184 | +func (e *metadataExporter) getAttrSet(ctx context.Context, keys []string) (attribute.Set, metadata.MD) { |
| 185 | + // Get each metadata key value, form the corresponding |
| 186 | + // attribute set for use as a map lookup key. |
| 187 | + info := client.FromContext(ctx) |
| 188 | + md := map[string][]string{} |
| 189 | + var attrs []attribute.KeyValue |
| 190 | + for _, k := range keys { |
| 191 | + // Lookup the value in the incoming metadata, copy it |
| 192 | + // into the outgoing metadata, and create a unique |
| 193 | + // value for the attributeSet. |
| 194 | + vs := info.Metadata.Get(k) |
| 195 | + md[k] = vs |
| 196 | + if len(vs) == 1 { |
| 197 | + attrs = append(attrs, attribute.String(k, vs[0])) |
| 198 | + } else { |
| 199 | + attrs = append(attrs, attribute.StringSlice(k, vs)) |
| 200 | + } |
| 201 | + } |
| 202 | + return attribute.NewSet(attrs...), metadata.MD(md) |
| 203 | +} |
0 commit comments