New Component: Encoding Extension

[MovieStoreGuy]

Is your feature request related to a problem? Please describe.
As one of the code owners for awskinesisexporter, and kafkaexporter, several issues get raised asking if we can support x encoding standard for the messages being written to these services.

My typical response that I am happy to accept any open data format standard but for any vendor specific encoding, it would need to be moved to a vendor specific exporter, ie: cloudwatchawskinesisexporter for cloudwatch specific encoding.

This is problematic because it makes onboarding new or existing encoding for given vendors (or even versioned encodings of currently supported ones) have a high barrier to entry and these vendor specific reimplement what is already existing code in some cases.

Describe the solution you'd like
What I am currently thinking is that the main opentelemetry-collector defines:

• An Encoding extension, to allow for supporting components (receivers, exporters) to know what they access to
• An Encoding factory that new encodings can be registered with on application start (same with how current components are registered)
• A httpreceiver and httpexporter that can accept the given encoding on a given endpoint, ie (reciever endpoint = /v1/zipkin/ingestion , encoding = otlpproto)
• An Encoding interface that defines version, name and Marshal* (I can't remember the exact name for the pdata marshaler's methods)

The reason being to include version is that this can be bumped as a side affect of dependency updates of a given release of the collector(contrib) which downstream systems could rely on being a fixed version.
(There was an open issue for kafkaexporter that was this case, I will need to find it to link it back)

Describe alternatives you've considered
Not sure any alternatives that would allow for reuse of encodings in packages but we could continue to have vendor specific exporters that extend base receivers.

Additional context

A nice little side affect of this approach is that vendor exporters could just be config alias with expected configuration.
Ie:

• influxdbexporter can just define the static fields in config that set the encoding to be one of influxdbproto, influxdbjson with a default exporter endpoint and security headers.

[atoulme]

It would be helpful for our use cases too. I support this move. I see it with azureeventhubreceiver too, we need to have the ability to reuse kafka codecs.

[MovieStoreGuy]

To help give a more technical overview and collect the ideas that I have, I will brain dump them here and start getting feedback:

Overall Goal

The encoding extension is to provide at its core:

• A standard means of sharing codecs between receivers and exporters
• Provide a means of version codecs to improve wire stability

This is to help ensure that across collector versions that a package upgrade doesn't break backwards compatibility when trying to produce/consumer data. (This is highly important for streaming applications like Kafka, Kinesis, Pulsar).

This should be considered safe enough to be part of core and contrib distributions of the collector.

Stretch Goals

I haven't looked at the feasibility with considerable detail, but the idea is:

• External supplied codecs

The idea here is that you could theoretically connect to something like AWS Glue or Kafka's schema registry where these codecs can be externally managed and implemented within the extension.

The concerns I have about feasibility here is:

• Performance
• Mappings

It has been request a few times for the Kafka exporter to provide a user supplied mapping from otlp to format of supplied choice which isn't possible today.

Outline

In order to implement this component, it would require the two things:

1. Definitions to be defined for codecs
2. An access pattern to pass versioned codecs to components

To achieve the first goal, a codec would have the following interface:

package codec

type Metric interface {
    pmetric.Marshaler
    pmetric.Unmarshaler
}

type Log interface {
    plog.Marshaler
    plog.Unmarshaler
}

type Trace interface {
    ptrace.Marshaler
    ptrace.Unmarshaler
}

The rationale for this approach is that these interfaces already exist and require minimal effort to implement the otlp codec, and it has been shown it is repeatable across the project with minimal code changes.

(Optional): An additional higher order type of codec to be defined that allows for length delimited encoding, ie:

message LengthDelimited {
    // Group if group has been set, the original data was broken up
    // and requires to be merged by `Order` value 
    int64 group = 1;
    // Count defines how messages are expected 
    // in order to generate the original value
    int64 count = 2;
    // Order defines where this message should be
    // placed in order to reconstruct the message
    int64 order = 3;
    // Data represents the actual codec that is being transported
    bytes data  = 4;
}

On the receiver side, it would mean if there was a group set, it would need to wait for some configured amount of time to reconstruct the original message but the logic would be:

func (rec *receiver) ProcessMessage(msg LengthDelimeted)
  if msg.group == 0 {
    rec.emit(rec.codec.Marshal(msg.data)) # The data didn't require to be split in order to transport
  }
  msgs = append(rec.group[msg.group], msg)
  if len(msg) != msg.count {
    rec.group[msg.group] = msgs
    return nil
  }
  sort.Sort(msgs, func(i, j int) bool {
    return msgs[i].order < msgs[j].order
  })
  var data bytes.Buffer
  for _, msg := range msgs {
     data.Write(msg.data)
  }
  rec.emit(rec.code.Marshal(data.Bytes()))
}

The above misses the idea of group by timeout but it is mostly the logic required for the receiver to join the values together again.

The rationale for why I'd like this is that it means that exporters that have fixed byte size limits (like Kafka and Kinesis) no longer have to bin pack each value in the marshalled codec format and potential drop values that are too big (which is becoming more and more likely with logging becoming more adopted).

A nice side affect of this is that we can remove the MarshalSizer logic from pdata since if something has a max allowed byte limit, it can use the higher order abstract of fixed byte limit encoder.

Furthermore, the reason why I want to define this in the codec's package instead of leveraging something that is more native to the service, is to reduce the complexity required in order to send data to it and keep it open to be used external from the collector to provide that flexibility.

The extension can be simply defined as:

package code

type Type struct {
   Name     string
   // Version defines the specific codec required
   // If the value is blank, then it is assumed to be the latest.
   Version  string
}

package encodingextension

type CodecFactory interface {
   // NewCodec creates a new codec type 
   // and returns an error if name and or version doesn't exist
   NewCodec(ct codec.Type) (codec.Codec, error)
}

The codecs that are accessible within the factory are those defined in pkg/codec (I believe this is where they currently exist).

Future impact

These are rather loose thoughts that I've come up with that could help extend the project further and reduce the potential toil experienced developing it.

Removal of vendor specific receivers or exporters to aliases

In a rather abstract terms of thinking, all receivers / exporters are effectively transport + codec.
The idea here is that we can provide a http, tcp, and udp receiver and exporter where it expects a codec (plus some extra config for auth and such).
Meaning if we have a codec for zipkin, jaeger, otlp, or vendor codec, we can very quickly have components that implement the required translation for service X without needed to recreate the existing transport with its respective lifecycles.

Now, this does mean it shifts left a lot of the configuration to the user since you now would have to pick transport + codec (+ plus any extras like auth and such). This is where the idea of an Component Alias would be handy since a you could define an alias for a specific vendor that abstracts the require config and works as if it were the original implemented component. ie:

---
receiver:
   chrony: # This could be an alias for what is defined below
      endpoint: unix:///var/run/chronyd.sock
      timeout: 10s
   udp:
      endpoint: unix:///var/run/chroynd.sock
      timeout: 10s
      socket_type: AF_UNIX
      codec:
         name: chrony

These would be considered equivalent in their implementation.

[aboguszewski-sumo]

@MovieStoreGuy this looks promising and we also would benefit from adding it. Few questions though:

• Will it be possible to define codecs as standalone components and just include them in an OTC distro, like it is currently done with other components? It seems as a good way to pass versioned codecs to components, but I don't know the internals of otelcol well yet.
• How do you envision the development of it? We currently want to add a custom marshaller to the AWS S3 exporter and we want to do this as soon as possible, so we will probably need to fork the component to do this (because adding it to the upstream wouldn't make much sense if such feature is coming). When do you think that, for example, a demo interface will be available so that we can write it in such way that the transition to this extension will be as harmless as possible? Additionally, we probably would be able to give some help in implementing the extension.
• Do you think it would be possible to chain codecs? For example in our case above, we would use a transport exporter, then s3 codec and our custom codec (in whatever appropriate order).

[MovieStoreGuy]

Will it be possible to define codecs as standalone components and just include them in an OTC distro, like it is currently done with other components?

Technically possible, since I imagine each would be their own package. Not to sure if the packages are public facing though. I should clarify when I say versioned codecs, I mean we are following how go intends you to my breaking versioning as defined by their Module release page, maybe not as strictly so the project doesn't need to release special tags for codecs but rather have a package have definitions of v1, v2, vX, etc.
Similar to the semconv package if you're familiar.

How do you envision the development of it?

My plan was to start it this week, and I have a draft in my local. Just trying to do all the wiring of it all.

Do you think it would be possible to chain codecs?

This makes me nervous, mostly because converting outside the internal format of the collector means some sort of dynamic adapter which I'd rather not do for the sake of keeping it simple.

The way I imagine components to consume the extension would look something like this:

extensions:
  encoding:
    # some time in the distance future, references remote schema sources

exporters:
  awss3:
    encoding: zipkin/v1-json
    compression: zflate

Might make sense to have a wire format field that is like protobuf, json, text, etc

[MovieStoreGuy]

Linking issue for codecs: #21152

[atoulme]

Do you need a sponsor @MovieStoreGuy ? Is core the right repo or is contrib easier?

[MovieStoreGuy]

It would nice to have another lead sponsor this to help keep the idea flowing.

For the time being, I think it makes sense in contrib to avoid code duplications.

[dmitryax]

I missed this proposal. It looks pretty promising and has some overlap with the work that I'm doing as part of open-telemetry/opentelemetry-collector#8122. I believe we can work independently up to introducing LengthDelimited. Then we should look into possible ways to converge.

[atoulme]

imo we're ways away from getting to LengthDelimited. There's a lot of work to get the extension adopted across our components, and specifically to reduce our reliance on the same marshalers and unmarshalers across multiple components right now.

[yurishkuro]

Clarification: do I understand correctly that this is not a new component type, just a regular extension? I.e. in order to use it another component would need to look up this extension by ID, and cast it to a specific interface, right?

[dmitryax]

Clarification: do I understand correctly that this is not a new component type, just a regular extension? I.e. in order to use it another component would need to look up this extension by ID, and cast it to a specific interface, right?

That's correct. We drifted from the design defined in this issue, though. The encoding now is a class of extensions implementing Marshal/Unmarshal, not the only extension. Components can point to any of those to get a specific format. The discussion to follow is in #27092.

[yurishkuro]

@MovieStoreGuy @dmitryax what about the startup order of extensions, is that supported somehow? E.g. if one uses a tap / websocket extension, those could also benefit from encoding extensions, but the encoding extensions would need to be initialized before others.

[dmitryax]

We never had any extensions depending on each other so far. The encoding extensions are meant to be used by the receivers and exporters. tap / websocket extension should be a processor AFAIK

[github-actions]

This issue has been inactive for 60 days. It will be closed in 60 days if there is no activity. To ping code owners by adding a component label, see Adding Labels via Comments, or if you are unsure of which component this issue relates to, please ping @open-telemetry/collector-contrib-triagers. If this issue is still relevant, please ping the code owners or leave a comment explaining why it is still relevant. Otherwise, please close it.

[github-actions]

This issue has been closed as inactive because it has been stale for 120 days with no activity.