apiVersion: batch/v1
kind: Job
metadata:
  name: kkvtest
spec:
  template:
    spec:
      shareProcessNamespace: true
      securityContext:
        runAsUser: 65534
        runAsGroup: 65534
        fsGroup: 65534
      hostAliases:
      - ip: "127.0.0.1"
        hostnames:
        - "kafka.local"
      - ip: "127.0.0.1"
        hostnames:
        - "hook.local"
      - ip: "127.0.0.1"
        hostnames:
        - "kkv.local"
      volumes:
      - name: redpanda-config
        emptyDir: {}
      - name: redpanda-data
        emptyDir: {}
      - name: logs
        emptyDir: {}
      containers:
      - name: kafka
        image: vectorized/redpanda:v21.9.1@sha256:fdf42728944df0a5e987bc4335b5618bd5f6aebf5fffe4dea28272112b4d16ed
        command:
        - /bin/bash
        - -cex
        - |
          rpk redpanda config bootstrap --id 0
          rpk \
            redpanda \
            start \
            --overprovisioned \
            --smp=1 \
            --memory=1G \
            --reserve-memory=0M \
            --check=false \
            --node-id=0 \
            --kafka-addr=PLAINTEXT://kafka.local:9092 \
            --advertise-kafka-addr=PLAINTEXT://kafka.local:9092
        ports:
        - containerPort: 9092
        volumeMounts:
        - mountPath: /var/lib/redpanda/data
          name: redpanda-data
        - mountPath: /etc/redpanda
          name: redpanda-config
      - name: topics
        image: vectorized/redpanda:v21.9.1@sha256:fdf42728944df0a5e987bc4335b5618bd5f6aebf5fffe4dea28272112b4d16ed
        command:
        - /bin/bash
        - -cex
        - |
          until rpk cluster --brokers kafka.local:9092 info; do sleep 1; done;
          # We seem to handle non-existent topics quite badly, ending up with too many unacked messages once consumption start
          # sleep 30
          rpk topic --brokers kafka.local:9092 create table -r 1 -p 1
          rpk topic --brokers kafka.local:9092 create onupdate -r 1 -p 1
      - name: hook
        image: docker.io/yolean/kafka-hook:1a21a5815dca637b5e88f0d78dfecf85d0276267@sha256:90931dbba5d15c7c86436dffcc5128ba25b0ffee35f2d99796184a2361161ad8
        args:
        - -Doutgoing.hook.bootstrap.servers=kafka.local:9092
        - -Doutgoing.hook.topic=onupdate
        - -Dcloudevent.source-host=https://kkvtest.yolean.se
        - -Dcloudevent.type-prefix=se.yolean.kkvtest.
        ports:
        - containerPort: 8080
        resources:
          requests:
            memory: "30Mi"
            cpu: "30m"
          limits:
            memory: "30Mi"
            cpu: "30m"
        readinessProbe:
          httpGet:
            path: /q/health/ready
            port: 8080
      # The legacy kkv image for comparison
      - name: kkv1
        image: docker.io/yolean/kafka-keyvalue:69608d815fbb525901b8acb6c315f9e39780a51b@sha256:6435695b3412d4ee37c7510c0f41cb4f16ff5add10030e8da4b67e9b83f385a7
        env:
        - name: QUARKUS_HTTP_PORT
          value: "8090"
        - name: topic
          value: table
        - name: kafka_bootstrap
          value: kafka.local:9092
        - name: kafka_group_id
          value: kkv001
        - name: kafka_offset_reset
          value: latest
        - name: target
          #value: http://hook.local:8080/hook/v1/kkv001
          value: http://hook.local:8089/hook/v1/kkv001
        ports:
        - name: http1
          containerPort: 8090
          protocol: TCP
      - name: kkv
        image: builds-registry.ystack.svc.cluster.local/yolean/kafka-keyvalue
        env:
        - name: QUARKUS_HTTP_PORT
          value: "8091"
        - name: topic
          value: table
        - name: kafka_bootstrap
          value: kafka.local:9092
        - name: kafka_group_id
          value: kkv002
        - name: kafka_offset_reset
          value: latest
        - name: target
          #value: http://hook.local:8080/hook/v1/kkv002
          value: http://hook.local:8089/hook/v1/kkv002
        ports:
        - name: http
          containerPort: 8091
          protocol: TCP
        readinessProbe:
          httpGet:
            path: /q/health/ready
            port: api
          timeoutSeconds: 2
          initialDelaySeconds: 30
        livenessProbe:
          httpGet:
            path: /q/health/live
            port: api
          timeoutSeconds: 2
          initialDelaySeconds: 30
      - name: test
        image: docker.io/yolean/toil:7a281dfc9eb329c14dd6e5814071a37885554f9c@sha256:68bc7443a42cc886dda1267fb95200cbc8026c2fe16ec4f178eabc2988413a3e
        env:
        - name: BOOTSTRAP
          value: kafka.local:9092
        - name: KKV1
          #value: http://kkv.local:8090
          value: ""
        - name: KKV
          value: http://kkv.local:8091
        - name: HOOK
          value: http://hook.local:8080
        - name: RQLOG
          value: http://hook.local:8089
        - name: RQLOGMETRICS
          value: http://hook.local:9901/stats/prometheus
        command:
        - /bin/bash
        - -cex
        - |
          set +e

          curl --retry 10 --retry-delay 1 --retry-connrefused $RQLOG/testsetup

          curl --retry 10 --retry-delay 3 --retry-connrefused $KKV/q/health/ready
          curl -s $RQLOGMETRICS | grep envoy_listener_http_downstream_rq_completed
          echo 'first' | kcat -b $BOOTSTRAP -P -t table -k t1 -e
          curl $KKV/cache/v1/keys | grep '"t1"'
          curl -s $RQLOGMETRICS | grep envoy_listener_http_downstream_rq_completed

          jq . /var/testlogs/requests.json
          # kkv1 is faster than redpanda up, bails on nonexisting topic, then restarts itself on the first readiness probe
          [ -z "$KKV1" ] ||curl $KKV1/q/health/ready
          [ -z "$KKV1" ] ||until curl -f -s $KKV1/cache/v1/raw/t1; do echo "Waiting for kkv1 to fill cache" && sleep 1; done
          echo 'second' | kcat -b $BOOTSTRAP -P -t table -k t2 -e
          [ -z "$KKV1" ] || curl -f -s $KKV1/cache/v1/raw/t2
          curl -f -s $KKV/cache/v1/raw/t2
          jq . /var/testlogs/requests.json

          N=1000
          seq 1 $N | kcat -b $BOOTSTRAP -P -t table -kt$N
          pkill -SIGSTOP redpanda
          RESULT=$(curl $KKV/cache/v1/raw/t$N)
          [ "$N" = "$RESULT" ] || echo "Expected $N got $RESULT"
          sleep 1
          RESULT=$(curl $KKV/cache/v1/raw/t$N)
          [ "$N" = "$RESULT" ] || echo "Expected $N got $RESULT"
          # Now test how kkv behaves after kafka went offline
          for i in {0..3}; do
            sleep 15
            [ -z "$KKV1" ] || curl $KKV1/q/health/ready
            curl $KKV/q/health/ready
            # Unlike first generation KKV we should expect response with the latest cached value during kafka downtime
            curl $KKV/cache/v1/keys
          done
          pkill -SIGCONT redpanda

          echo "Onupdate requests logged so far: $(wc -l /var/testlogs/requests.json)"

          echo "Restarting kkv"
          curl -X POST $KKV/_admin/v1/shutdown/199
          seq 1 10 | kcat -b $BOOTSTRAP -P -t table -k post-shutdown

          tail -n 1 /var/testlogs/requests.json | jq '.'
          curl --retry 10 --retry-delay 3 --retry-connrefused -f $KKV/q/health/ready
          seq 1 10 | kcat -b $BOOTSTRAP -P -t table -k post-shutdown-ready
          curl $KKV/cache/v1/raw/post-shutdown
          curl $KKV/cache/v1/raw/post-shutdown-ready
          tail -n 1 /var/testlogs/requests.json | jq '.'
          sleep 1
          curl $KKV/cache/v1/raw/post-shutdown
          curl $KKV/cache/v1/raw/post-shutdown-ready
          tail -n 1 /var/testlogs/requests.json | jq '.'

          echo "Verifying the very basic case"
          date -Ins | kcat -b $BOOTSTRAP -P -t table -k t9
          last=$(kcat -b $BOOTSTRAP -C -t table -f '%o\n' -o -1 -e)
          sleep 1
          tail -n 2 /var/testlogs/requests.json | grep kkv002 | grep $last
          tail -n 2 /var/testlogs/requests.json | grep kkv001 | grep $last
          sleep 1
          tail -n 2 /var/testlogs/requests.json | grep kkv002 | grep $last
          tail -n 2 /var/testlogs/requests.json | grep kkv001 | grep $last
          sleep 1
          tail -n 2 /var/testlogs/requests.json | grep kkv002 | grep $last
          tail -n 2 /var/testlogs/requests.json | grep kkv001 | grep $last

          tail -f /dev/null
        volumeMounts:
        - name: logs
          mountPath: /var/testlogs
      - name: requestlog
        image: envoyproxy/envoy:v1.19.1@sha256:ac6a29af5bee160a1b4425d7c7a41a4d8a08a7f9dd7f225f21b5375f6439457a
        ports:
        - containerPort: 8089
          name: http
        - containerPort: 9901
          name: metrics
        volumeMounts:
        - name: logs
          mountPath: /var/testlogs
        command:
        - /usr/local/bin/envoy
        - -l
        -   info
        - --config-yaml
        - |

          admin:
            access_log_path: /dev/null
            address:
              socket_address:
                protocol: TCP
                address: 0.0.0.0
                port_value: 9901
          static_resources:
            listeners:
            - name: listener
              address:
                socket_address:
                  address: 0.0.0.0
                  port_value: 8089
              filter_chains:
              - filters:
                - name: envoy.filters.network.http_connection_manager
                  typed_config:
                    "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
                    stat_prefix: log
                    access_log:
                    - name: envoy.access_loggers.file
                      filter:
                        not_health_check_filter:  {}
                      typed_config:
                        "@type": type.googleapis.com/envoy.extensions.access_loggers.file.v3.FileAccessLog
                        path: "/var/testlogs/requests.json"
                        log_format:
                          json_format:
                            start_time: "%START_TIME%"
                            req_method: "%REQ(:METHOD)%"
                            req_path: "%REQ(X-ENVOY-ORIGINAL-PATH?:PATH)%"
                            bytes_recv: "%BYTES_RECEIVED%"
                            agent: "%REQ(USER-AGENT)%"
                            req_id: "%REQ(X-REQUEST-ID)%"
                            req_content_type: "%REQ(CONTENT-TYPE)%"
                            resp_code: "%RESPONSE_CODE%"
                            req_topic: "%REQ(X-KKV-TOPIC)%"
                            req_offsets: "%REQ(X-KKV-OFFSETS)%"
                    route_config:
                      name: dummy
                      virtual_hosts:
                      - name: dummy
                        domains: ["*"]
                        routes:
                        - match:
                            prefix: "/"
                          direct_response:
                            status: 200
                            body:
                              inline_string: >
                                logged
                    http_filters:
                    - name: envoy.filters.http.router
      # We might want to switch to OnError as that's quite similar to a separate pods setup
      restartPolicy: OnFailure
