Only support the Kafka storage in Collector

[ledor473]

Which problem is this PR solving?

• Solves [ingester] Crash if Kafka is specified as storage #1190

Short description of the changes

• Add an unsupported list to the storage factory
• Make Kafka as unsupported in Ingester, Query, all-in-one (because Ingester is not started anyway)

[yurishkuro]

if we want to introduce optional params, we should switch to passing an Options struct instead of positional args.

https://github.com/jaegertracing/jaeger-lib/blob/3fa9aa04b65a266a1bbe0a5c8ac1408281b2a64a/metrics/factory.go#L31

[ledor473]

I choose to make it an optional parameter as the FactoryConfig is currently only for config from the Cobra, but I could indeed add an Options struct inside the same package and use that!

[codecov]

Codecov Report

Merging #1231 into master will not change coverage.
The diff coverage is 100%.

@@          Coverage Diff           @@
##           master   #1231   +/-   ##
======================================
  Coverage     100%    100%           
======================================
  Files         159     159           
  Lines        7145    7152    +7     
======================================
+ Hits         7145    7152    +7

Impacted Files	Coverage Δ
plugin/storage/factory.go	100% <100%> (ø)	⬆️
plugin/storage/factory_config.go	100% <100%> (ø)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 5f6af99...80b6a82. Read the comment docs.

[yurishkuro]

The design of this fix doesn't sit well with me. Right now the details of the storage are fully abstracted from the binaries, but here we're introducing some very specific & peculiar dependency that they have to declare Kafka as unsuitable storage.

The reason Kafka doesn't work for all-in-one is because it does not implement the SpanReader. The reason it does not work for ingester is because the flags are clashing (aside from the flags issue, however, there's no strong reason why ingester wouldn't do Kafka->Kafka, e.g. as a form of replicating data to another topic). If we somehow solved the clash of flags issue, then I assume all-in-one would've exited with a better error message like "reader not implemented", and we wouldn't need to do any changes. Alternatively, if we did want to handle it more explicitly, then the storage factory could expose metadata methods like CanRead(), CanWrite(), and all-in-one would check both and fail for Kafka storage whose factory would return false from CanRead().

[ledor473]

there's no strong reason why ingester wouldn't do Kafka->Kafka, e.g. as a form of replicating data to another topic)

@yurishkuro if we want to support that, maybe we could have a set of flags for the consumer and another set of flags for the producer. Something like this:

--kafka.consumer.brokers
--kafka.consumer.topic
--kafka.consumer.encoding
--kafka.consumer.group-id

And:

--kafka.producer.brokers
--kafka.producer.topic
--kafka.producer.encoding

This way you can both replicate from one topic to another, but we can also replicate to one cluster to another cluster. This would also get around the flags being declared twice.

[yurishkuro]

@ledor473 do you plan to finish this one?

[ledor473]

@yurishkuro I was waiting for a confirmation on the latest suggestion (renaming the flags to avoid the exception and to allow the use of multiples brokers/topics).

I might have missed the 👍 on it, but yes I plan to fix this PR.

[yurishkuro]

cool.

I think GitHub doesn't notify of reactions via +1 button, I'll make sure to comment explicitly in the future.

[ledor473]

@yurishkuro the change is mostly done in another branch but I see that we do not pass the logger to InitFromViper (https://github.com/jaegertracing/jaeger/blob/master/plugin/storage/kafka/options.go#L75-L81).

I see 2 options to log the current flag as deprecated:

1. I change that method in the Configurable interface to provide a logger and use logger.warn

   • Similar to this:

     jaeger/cmd/agent/app/reporter/tchannel/flags.go

     Lines 70 to 74 in 80b3dbb

     	func (b *Builder) InitFromViper(v *viper.Viper, logger *zap.Logger) *Builder {
     	if len(v.GetString(collectorHostPort)) > 0 {
     	logger.Warn("Using deprecated configuration", zap.String("option", collectorHostPort))
     	b.CollectorHostPorts = strings.Split(v.GetString(collectorHostPort), ",")
     	}

2. I use fmt.Printf

   • Similar to this:

     jaeger/plugin/storage/factory_config.go

     Lines 88 to 93 in 80b3dbb

     	fmt.Fprintf(
     	log,
     	"WARNING: found deprecated command line option %s, please use environment variable %s instead\n",
     	token,
     	SpanStorageTypeEnvVar,
     	)

[yurishkuro]

I would go with option 2, much fewer changes this way

[ledor473]

#1360 is merged, closing this one