Strimzi-kafka-operator: Create Connector operator

[jcechace]

We need an operator for deploying Kafka Connect connectors.

This issue should also serve as a place to gather the initial requirements for the operator.

[renatomefi]

This is related to #1076

[gwkunze]

Some initial thoughts:

• Using a labelSelector in a Kafka Connect CRD to associate Connectors to Kafka Connect instances would probably be the most idiomatic way of linking the two.
• Most of the configuration schema of individual Connectors is left up to the Connector code. There are very few fixed configuration keys. I think the differences between Sink and Source Connector configurations aren't significant enough to warrant separate CRDs for each.
  
  
  
  • It would probably be nice to support the status sub-resource to expose the status of the connector.
  
  
  • Validation that @gunnarmorling mentions in #1076 would be nice but could be potentially harmful if new configuration options are implemented in a connector that the CRD/Operator doesn't know about. Maybe a relaxed validation mode or a separate implementation as an adminssion controller might be an option there.
  
  

[tombentley]

I don't think trying to link some KafkaConnector CRs with a KafkaConnect is really the right approach. To me an approach that's similar to the relationship between a StatefulSet and its Pods would be more appropriate. In other words we would introduce a new CR, let's all it KafkaConnectors. It might look like this:

kind: KafkaConnectors
spec:
  replicas: 3
  template:
    connectors:
      - connector: 
          mavenSource: 
            // this is just a way of specifying the connector jar
            // and its transitive deps
            groupId: ...
            artifactId: ...
            version: ...
          config:
            // the connector.properties
          livenessProbe:
            // a probe for checking liveness of this connector
          readinessProbe:
            // a proble for checking readiness of this connector
    // also some way to configure some details of the deployment and pods,
    // for example labels and annotations.

Maybe there are other ways of specifying the source for the connector jar files. For example on OpenShift there could be some kind of S2I support for building the jars into an image. But for the mavenSource the Deployment orchestrated by the operator would be responsible for fetching the artifacts from a maven repo. This means for mavenSource the Docker image isn't really an immutable record of all the software that's in the container since it does not include the jars. But, of course, a properly run maven repo would be immutable, so it's not like it's not impossible to reproduce the exact same deployment, it's just that the image doesn't provide that property on its own.

Such CRs would be consumed by an operator, which could be configured with, for example:

• the maven repo(s) used to resolve the mavenSource (i.e. they don't have to come from central).
• perhaps some way to validate the config for individual connectors (though it's difficult to see how that would work if there's not a way to identify connectors; maven coord work for maven sources, but if there are other sources...).

The operator would orchestrate a Deployment. The pods for the Deployment would spin up connect-distributed.sh. The operator would then make the appropriate REST calls to spin up the given connector(s). The operator would also be responsible for monitoring the health of each connector via the probes, and restarting connectors if necessary (see #1479 for where this requirement is coming from). The operator could publish the state of each connector via the CR status subresource.

Obviously this is very much a sketch (I've not thought about propagating things like secrets to the connectors, for example), but @gunnarmorling @jcechace is this the sort of thing you envisaged, or am I way off track here?

[gwkunze]

@tombentley I'm not completely clear on what you intend, do you mean that this KafkaConnectors would replace the current KafkaConnect CRD and both run Kafka Connect, install connectors and use the Kafka Connect Rest api to run them?

[tombentley]

I didn't say replace, but what I proposed would be an alternative to using KafkaConnect and would alleviate users of the CR from having to interact with the REST API at all (at least in order to get connectors deployed), because the operator would do that itself. But really it's a bit of a straw man, so feel free to tear it apart or make suggestions.

[gunnarmorling]

What's the reasoning for making this a new CRD type instead of just adding this config to the KafkaConnect CRD itself? So it would allow you to deploy a KC cluster with a set of given connectors?

Regarding liveness/readiness probes: is there need for making this customizable per connector? Would it (at least initially) suffice perhaps to simply obtain the connector/task status via the KC REST API? That could be done in a connector-agnostic way.

[jcechace]

Personally, I'd rather see a CR per connect (e.g KafkaConnector) instead of the composite type @tombentley. If some configuration should be common for all connector then it should perhaps be part of the KafkaConnect CR.

As for the potential s2i and jar/maven sources -- that would be the next step (as well as config validation)

@gunnarmorling Initially that in combination with some configurable delay in between connector deployments. As @gwkunze mentioned during the call the potential redistribution of connectors might interrupt an ongoing initialisation phase of some connector (e.g. snapshotting in Debezium).

[tombentley]

What's the reasoning for making this a new CRD type instead of just adding this config to the KafkaConnect CRD itself? So it would allow you to deploy a KC cluster with a set of given connectors?

That would surely _work_, but I suppose (on reflection) it would leave a _mixed model_ where connectors could be created by both by declaration and by using the REST endpoint. It destroys the point of these CRs being declarative if they can be configured after the fact. OTOH it means that we'd have to hide the REST API away so people couldn't access it, which seems like a bit of effort for no additional flexibility.

Personally, I'd rather see a CR per connect (e.g KafkaConnector) instead of the composite type @tombentley. If some configuration should be common for all connector then it should perhaps be part of the KafkaConnect CR.

I think that works in a model where a KafkaConnect is linked/referenced by zero or more KafkaConnector instances (my straw-man was a KafkaConnectors and operator which did both of those things together). I don't think we should lock ourselves into a one-connector per connect deployment model though.

[gwkunze]

Apologies for the late response. One of the reasons I proposed a CR per Connector is the separation of operational concern.

Deploying KafkaConnect (besides it already being done by Strimzi) and deploying individual connectors might be done by different parties (Kafka administrators vs developers depending on your organization). Having the ability to have separate RBAC rules for both resources could benefit people there.

With the KafkaConnector as a separate resource would make it possible to make a KafkaConnector Operator which would essentially manage the REST api of a Kafka Connect deployment. Just like the Topic Operator and the User Operator these could then also be used by people running Kafka Connect manually in- or outside Kubernetes/OpenShift.

[jcechace]

I think that works in a model where a KafkaConnect is linked/referenced by zero or more KafkaConnector instances (my straw-man was a KafkaConnectors and operator which did both of those things together). I don't think we should lock ourselves into a one-connector per connect deployment model though

@tombentley ... I actually meant a separate CR per Connector not per Connect...

[scholzj]

I would tend to agree that separate CR per connector linked through label to the Connect API would be a good start. It would better capture the Connect architecture and make it easier to track the status of the different connectors. It would be also straight forward to use without the rest of Strimzi with any Kafka Connect.

However, it would make it a bit harder to add later the build part - to have the connector added tot he container images just based on the specification in the Connector resource, because that would need to be integrate with the part deploying the Connect cluster it self, not just the REST API.

[ppatierno]

If the REST API of the Kafka Connect remains still accessible by the users, it allows them to create new connectors without using the new KafkaConnector custom resource, which means a difference between the real status of the Kafka Connect workload and the reflected Kubernetes world (so more connectors than the KafkaConnector(s)). It's like user can still create a topic using the Kafka tool but then the TO reconcile creating the KafkaTopic resource but I would avoid that, hiding the REST API to users. Wdyt?

[scholzj]

I would do the same ... I would use it to secure the API basically. But I guess that might be only optional (or a feature of how Strimzi deploys it ;-)) since that might not be the preferred way for everyone. But we should definitely avoid any bi-directional sync ... I think we have enough of that with topics :-o.

[jcechace]

Speaking of using the operator without the rest of Strimzi... If we decide to accept that as a valid use case (for us) then I'd actually propose to decouple the operator from rest of the Strimzi code base -- and ideally write it in go using the standard operator framework. WDYT?

[scholzj]

I'm not necessarily against it - but I do not think it is related. The User or Topic operators can be also used independently on whether the Kafka cluster is deployed by Strimzi or not. Also, I'm not sure how useful would the operator framework be since this will not be totally traditional operator - it will have a CRDs on one end, but the other end will be basically a custom HTTP service and not Kubernetes API. So you should check how well does the Operator Framework fit from this perspective.

[pamelipluas]

@jcechace hey! I want to know if you have any progress with this issue? I'm using the library and we'd love to have the connector operator functionality

[pamelipluas]

@tombentley @scholzj @ppatierno hey guys, we (with @faustocv) are intended to develop this operator and collaborate with the project. We'd like to have a conversation with you in order to understand the current resources.

[scholzj]

Hi @pamelipluas ... I was planning to look into this during one of the comming weekends. But we always welcome new contributors, so if you want to work on this it would be great and I will be happy to leave it for you and @faustocv (and help you with it where needed). Are you on our Slack channel? Maybe we can sync there and discuss the next steps (either directly there or just plan some conference call or somehting).

[pamelipluas]

Some starting code related to the Custom Resource Model. Feedback is welcome.
https://github.com/pamelipluas/strimzi-kafka-operator/commit/670d2d68c12d73d7b2fe1bd564ca804de55c6b8b
https://github.com/pamelipluas/strimzi-kafka-operator/commit/137db90a0040ecceb66c64d7afb4621ea479d845

cc @jcechace @tombentley

[tombentley]

Or viewed as a diff https://github.com/strimzi/strimzi-kafka-operator/compare/master...pamelipluas:master

[tombentley]

I've created a topic branch in the strimzi/strimzi-kafka-operator repo for this and a PR https://github.com/strimzi/strimzi-kafka-operator/pull/2001. This will make it easier to review these changes and also mean you can open future PRs for the overall feature against that feature branch @pamelipluas.

[tombentley]

@pamelipluas has been making good progress on this, but I've come across a wrinkle that might lead us to reconsider the proposed design.

Unlike most of the rest of the operators in Strimzi (except for the KafkaUserOperator) this one needs to take care of CR deletion to make a DELELTE request to delete a connector. In particular we need to ensure that the operator will delete connectors which were deleted while the operator was not running. For the other operators this is not a problem because either:

• They can rely on Kubernetes GC (because the derived resources are all other Kubernetes resources)
• Or in the case of the KafkaUserOperator it knows about the Kafka cluster and because there's never > 1 UO for a Kafka cluster it is able to query Kafka to find out about all the users it knows about and infer from that which KafkaUser resources must have been deleted.

Neither of those patterns applies here. The problem arises because the information identifying the Kafka Connect cluster to delete the connector from resides in the KafkaConnector resource itself. I.e. the information is lost when the resource is deleted, so the operator doesn't know which KafkaConnect cluster to query for connectors.

A solution to this would be to reverse how we represent the parent-child relationship between the KafkaConnect and the KafkaConnectors which should run within it. Specifically, if the KafkaConnect used a selector to identify connectors which should run within it, then even if a KafkaConnector was deleted while the operator was not running there would still be enough information available in Kubernetes to know which connectors should be deleted.

So concretely an example KafkaConnector resource might look like:

apiVersion: "kafka.strimzi.io/v1alpha1"
kind: "KafkaConnector"
metadata:
  name: "test-source"
  labels:
    connect-cluster: foo
spec:
  class: "org.apache.kafka.connect.file.FileStreamSourceConnector"
  tasksMax: 2
  topics:
  - "test.topic"
  config:
  - name: "file"
    value: "/home/source/test.source.txt"

(note the connector-cluster: foo).

And the KafkaConnect CR would look something like this:

apiVersion: "kafka.strimzi.io/v1beta1"
kind: "KafkaConnect"
metadata:
  name: "test-kafka-connect"
spec:
  replicas: 6
  image: "foo"
  config:
    name: "bar"
  bootstrapServers: "kafka:9092"
  selector:
    connect-cluster: foo

There's nothing stopping multiple KafkaConnect resources having the same selector. Conceptually that is not a problem (it is not an error to have two connect clusters running the same connector), but it complicates things when it comes to the reporting of status on the KafkaConnector. For this reason we might want a way to prevent multiple parentage.

Can anyone see an alternative? Or suggest improvements on what I've sketched out here?

[tombentley]

Is there much value in having a way to pause a connector?

[tombentley]

By which I mean, a way to pause a connector _via the CR_.