smartagent/collectd/custom not working for custom collectd python plugin script

[vathanaka]

We are trying to migrate from SFX agent to new Splunk Otel Collect chart using helm chart.

We have a custom/collectd plugin script written in python and want to migrate it, however the custom/collect python script not working.

Can some one help to correct my config/script?

values.yaml

nameOverride: "splunk-otel-agent"
clusterName: "private-cluster"
splunkObservability:
  realm: "us0"
  accessToken: "CHANGEME"
cloudProvider: "gcp"
distribution: "gke"
environment: staging
readinessProbe:
  initialDelaySeconds: 120
livenessProbe:
  initialDelaySeconds: 120
tolerations:
  - effect: NoSchedule
    key: message-generator-node
    operator: Equal
    value: allowed
clusterReceiver:
  enabled:
logsCollection:
  containers:
    excludeAgentLogs: false
    enabled: true
agent:
  extraVolumeMounts:
    - name: kafka-topic-monitor
      mountPath: /mnt/custom-plugins-3/
  extraVolumes:
    - name: kafka-topic-monitor
      configMap:
        name: kafka-topic-monitor
  config:
    extensions:
      k8s_observer:
        auth_type: serviceAccount
        node: ${K8S_NODE_NAME}
      smartagent:
        collectd:
          logLevel: info
          readThreads: 5
          timeout: 40
          writeQueueLimitHigh: 500000
          writeQueueLimitLow: 400000
    receivers:
      receiver_creator:
        receivers:
          smartagent/collectd/custom/kafka-topic-monitor:
            config:
               type: collectd/custom
            rule: type == "port" && port == 8080 && pod.labels["name"] == "message-generator"
            template: |
                LoadPlugin python
                TypesDB "/mnt/custom-plugins-3/types.db"
                <Plugin python>
                  ModulePath "/mnt/custom-plugins-3/"
                  LogTraces true
                  Import "kafka-topic-monitor2"
                  <Module "kafka-topic-monitor2">
                    Host "{{.Host}}"
                    Port "{{.Port}}"
                  </Module>
                </Plugin>
        watch_observers:
        - k8s_observer
    processors:
      metricstransform:
        transforms:
        - action: update
          include: .*
          match_type: regexp
          operations:
          - action: add_label
            new_label: kubernetes_cluster
            new_value: staging-cluster
      resourcedetection:
        detectors:
        - system
        - env
        - gcp
    exporters:
      signalfx:
        access_token: ${SPLUNK_OBSERVABILITY_ACCESS_TOKEN}
    service:
      telemetry:
        logs:
          level: debug
      pipelines:
        metrics:
          receivers:
          - receiver_creator
          exporters:
          - signalfx
      extensions:
      - k8s_observer
      - smartagent

python script in k8s's configmap

apiVersion: v1
data:
  kafka-topic-monitor2.py: |
    from math import pow
    import collectd
    import urllib.request, urllib.error, urllib.parse
    import json
    import re

    PLUGIN_NAME = 'kafka-topic-monitor2'
    INTERVAL = 30  # seconds
    _host = None
    _port = None

    collectd.info('kafka-topic plugin: Loading Python plugin:' + PLUGIN_NAME)

    def configure(config_obj):
        '''Example configuration:

        - type: collectd/custom
            template: |
              LoadPlugin "python"
              <Plugin python>
                ModulePath "/usr/lib/splunk-otel-collector/agent-bundle/3.8"
                Import "kafka_topic_monitor2"
                <Module "kafka_topic_monitor2">
                    Interval "5"
                </Module>
              </Plugin>
        '''

        global _host
        global _port

        config = {c.key: c.values[0] for c in config_obj.children}

        _host = config.get('Host')           or 'localhost'
        _port = config.get('Port')           or 8080


    def request(endpoint):
        collectd.info('kafka-topic-monitor plugin: send request...')
        url = 'http://' + _host + ':' + str(_port) + endpoint
        req = urllib.request.Request(url)
        response = urllib.request.urlopen(req) # Throw on failure
        return response.read()


    def get_topics(env):
        collectd.info('kafka-topic-monitor plugin: get topics...')
        res = request('/api/environment/' + env + '/topics')
        parsed = json.loads(res)
        return parsed


    def get_queue_length(topic):
        collectd.info('kafka-topic-monitor plugin: get queue length %s' % topic)
        res = request('/api/topic/' + topic + '/queued')
        parsed = json.loads(res)
        return parsed.get('no_queued_message')


    def format_dimensions(topic, namespace):
        collectd.info('kafka-topic-monitor plugin: format dimesions')
        data = {
            'topic': topic,
            'namespace': namespace
        }
        dim_pairs = ["%s=%s" % (k, v) for k, v in data.items()]
        return "[%s]" % (",".join(dim_pairs))


    def read(data=None):
        prod = get_topics('production')
        collectd.info('kafka-topic-monitor plugin: get topics from prod %s' % prod)

        for topic in prod:
            val = collectd.Values(type='gauge', type_instance="gauge.kafka.queue_length")
            val.plugin = PLUGIN_NAME
            val.values = (get_queue_length(topic),)
            val.plugin_instance = format_dimensions(topic, "kafka")
            val.dispatch()


    collectd.register_config(configure)
    collectd.register_read(read, INTERVAL)
kind: ConfigMap
metadata:
  name: kafka-topic-monitor2