Access Strimzi kafka through python

[iamrsaravana]

We are planning to create dynamic topics, connect, connector through python program. In order to access through code, we are getting ip address, token, certificate from the below service account tokens:

(currently cluster strimzi kafka is running in GKE cluster)

Using below commands to get token and certificates

CAcertifcates:
kubectl -n kafkanew get secrets my-cluster-entity-operator-token-9cst7 -o jsonpath="{['data']['ca.crt']}"

Token:
kubectl -n kafkanew get secrets my-cluster-entity-operator-token-9cst7 -o jsonpath={.data.token} | base64 -d

Python code for creating topics:

from kubernetes import client
import base64
from tempfile import NamedTemporaryFile
import os
import yaml
from os import path

def main():
try:
host_url = 'https://35.224.26.16'
cacert = 'LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0tCk1JSURLekNDQWhPZ0F3SUJBZ0lSQU1pSHRSSTY5Tm95ZFmNtemhjQmJtUGJCNFhNUGs3SFZoNEZDdUhBCllQdDV2eUpLVTBHSnpJeWUybjM1SlVQc1ljR1J1Yk5JajduNEM4RWg5L2NyK3htVGUwVDRDaE83NUVUckl5WkgKT3dtMnRaZVR4TVhOR0h3RnRRTUZZbEtoU20rb2tCekErRW5uZ1JlRTh6c2hEMEUvMUVRN0tUWExKNEdUUHdYVwpzUG5iNHRWREtXTWVaUHhNYnkwOFp1bHJQRWcyVjZvUFd6TjA4cyszRVVhMnB1L041dHhXcE1Yd0MxbTFDd3htCmFMZ05kekRZakd5UHQ5S0FpTUZKOGN1RDFOaEpQdUd3S0RWR3BnL2Q3ZUVaMFFJREFRQUJvMEl3UURBT0JnTlYKSFE4QkFmOEVCQU1DQWdRd0R3WURWUjBUQVFIL0JBVXdBd0VCL3pBZEJnTlZIUTRFRmdRVURqSGxHeTk3RVg5MwphT0ZOT3E1UzJrR0VQcEV3RFFZSktvWklodmNOQVFFTEJRQURnZ0VCQUtMeHJWQUVMMkRkakM0ZTFzaVl5VUxvCmtsMDJpVEFyV050NDE3eTBhVDlSSk0reVdmZUVSZjhnaWd6L3VSODdnZzdjTjVvT2RDSkVyS1ZscXozeXVBTnEKZ3pkcDZiOENZYjRUbGtNaElWbnB2ZHg2MUQyS3MxdDBGNWZsejhQdXNZUjRRYW1GSUFhY1FteXpBdW9uWHQ0RgpTN3VrUmN1ODNKS1dyUkRnekxyMjJIbVVUeTZEMCs5YU44Wkl0SWoyR051NDVaSmhTbHZYZ3JQS2o4bEJ4YXROClk1Nm8wU0pmRHNqVlZsckRPUFZpc1ZEdjl0TmhiamdRdVJNZTJKQXJuZ1dVbkZiVkZzUHlYS3VvckRuWUVJWG8KQmVBM3l2YzlVdDdmWEJlNklsNnJLcDE0YzRsYlU4OVM0ZzNIb1dzdlhWVk9NM1BUdjVQbVpMQXRNTnowcjUwPQotLS0tLUVORCBDRVJUSUZJQ0FURS0tLS0tCg=='
token = 'eyJhbGciOiJSUzI1NiIsImtpZCI6IlZ6Z29wZTg3OUF3ZjJHNnotQldrODFmUWRDMHgzeEVpWWlISWV1Q2cwTUEifQ. -iGqp8p0hUS2IbX1bNiejfWcA6VLoL0F773ADN3DNu_Hn0NBFUZv7pCucqTlJK4TToYVC4-tTvFNCo7F7y9qDczENnQs9kqWpOmBiV9e-aicSKPLMI_5dkAphXpab91rtzZ0CAIpvhDWhK7QS1m1EQ9YSK5oK108r3T16rXRPENN1XM5HXsLXZ7-4AF52k_KKcxZJdluypXsnl3LbzoXx5jWaGPcry_erv5bek9sRraHkQw5FMTp2ljuUNgSveJtYnzg'

    print('read parameters')
    # Set the configuration
    configuration = client.Configuration()
    with NamedTemporaryFile(delete=False) as cert:
        cert.write(base64.b64decode(cacert))
        configuration.ssl_ca_cert = cert.name
    configuration.host = host_url
    configuration.verify_ssl = True
    configuration.debug = False
    configuration.api_key = {"authorization": "Bearer " + token}
    client.Configuration.set_default(configuration)
    print('read all certificates')
    # Prepare all the required properties in order to run the create_namespaced_job method
    # https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/BatchV1Api.md#create_namespaced_job
    v1 = client.BatchV1Api()
    with open(path.join(path.dirname(__file__), "topic.yaml")) as f:
        body = yaml.safe_load(f)
    print('try to apply ymal file')
    print('body'+str(body))
    v1.create_namespaced_job(namespace="kafkanew", body=body, pretty=True)

    return f'Job created successfully', 200

except Exception as e:
    print('some error in code'+str(e))
    return str(e), 500

if name == 'main':
main()

topic yaml file:
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
name: my-pythontesttopic
labels:
strimzi.io/cluster: my-cluster
spec:
partitions: 3
replicas: 3
config:
retention.ms: 7200000
segment.bytes: 1073741824

while executing above python code I am getting below error message:

read parameters
read all certificates
try to apply ymal file
body{'apiVersion': 'kafka.strimzi.io/v1beta2', 'kind': 'KafkaTopic', 'metadata': {'name': 'my-pythontesttopic', 'labels': {'strimzi.io/cluster': 'my-cluster'}}, 'spec': {'partitions': 3, 'replicas': 3, 'config': {'retention.ms': 7200000, 'segment.bytes': 1073741824}}}
some error in code(403)
Reason: Forbidden
HTTP response headers: HTTPHeaderDict({'Audit-Id': 'b44c7f92-7d24-41b8-93ae-6caecdd74ebf', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Content-Type-Options': 'nosniff', 'Date': 'Thu, 19 Aug 2021 11:28:06 GMT', 'Content-Length': '387'})
HTTP response body: {
"kind": "Status",
"apiVersion": "v1",
"metadata": {

},
"status": "Failure",
"message": "jobs.batch is forbidden: User "system:serviceaccount:kafkanew:my-cluster-entity-operator" cannot create resource "jobs" in API group "batch" in the namespace "kafkanew"",
"reason": "Forbidden",
"details": {
"group": "batch",
"kind": "jobs"
},
"code": 403
}

Could you suggest how to create/modify strimzi kafka through Kubernetes API. Is there special role I need to create it? in order to create topic, connector and connect through python?

[scholzj]

I have no idea how to do it in Python. In Java we have an api JAR to help with it. In Golang with the official Kubernetes client, you could use what IIRC is called Raw resources. But i have no idea about Python. I would suspect that the BatchV1 API is for creating Jobs and not for creating topics. Maybe you can check the source codes of the https://github.com/systemcraftsman/strimzi-kafka-cli which is IMHO written in Python and must do similar things.

But what I can do, is I can tell you about the error ... you should not use the entity operator service account. You should create your own service account for it with your own rights. The error you have here is basically just saying that with the entity operator service account you cannot create a Job resource because of missing RBAC (and because as I said above, the BatchV1API will create a Job and not a KafkaTopic).

Also, keep in mind that the Topic Operator is special in that it works in both directions. So you can also just create the topic in Kafka and it will create the KafkaTopic resource itself.

[iamrsaravana]

Thanks a lot for your reply,

we have modified "v1=client.APPsV1Api" instead of "v1 = client.BatchV1Api()". now it is passing as Kafka topic instead of jobs.

Current python code is not accepting CRD related resources. we are getting below error message

I have gone through Strimzi-kafka-cli python package, which is downloading kubectl local and executing command. we are planning to call this python code from cloud function. looks like it is not supporting remote kubernetes cluster.

[iamrsaravana]

could you share where can i get Java Jar file to accomplish this.

[scholzj]

Well, the api module is available in Maven repositories: https://mvnrepository.com/artifact/io.strimzi/api ... so you can get it there. But you cannot use it from Python of course, you would need to write your application in Java of course. I have one example here: https://github.com/scholzj/strimzi-api-examples ... but it is not really creating topics (still might give you an idea).