#!/bin/bash -e
#
# Apache v2 license
# Copyright (C) 2023 Intel Corporation
# SPDX-License-Identifier: Apache-2.0
#

# General setting
DEFAULT_OR_GATED=${1:-default}
WORKLOAD=${WORKLOAD:-kafka}
BACKEND=${BACKEND:-kubernetes}
TIMEOUT=${TIMEOUT:-3000}

# Cluster Setting
BROKER_SERVER_NUM=${BROKER_SERVER_NUM:-1}
PRODUCER_NODE_NUM=${PRODUCER_NODE_NUM:-1}
CONSUMER_NODE_NUM=${CONSUMER_NODE_NUM:-1}
TOPIC_NUM=${TOPIC_NUM:-1}
# Kafka Setting
REPLICATION_FACTOR=${REPLICATION_FACTOR:-1}
KAFKA_BENCHMARK_TOPIC=${KAFKA_BENCHMARK_TOPIC:-KAFKABENCHMARK}
MESSAGES=${MESSAGES:-10000000}
NUM_RECORDS=${NUM_RECORDS:-5000000}
THROUGHPUT=${THROUGHPUT:--1}
RECORD_SIZE=${RECORD_SIZE:-1000}
COMPRESSION_TYPE=${COMPRESSION_TYPE:-lz4}
CONSUMER_TIMEOUT=${CONSUMER_TIMEOUT:-600000}
BUFFER_MEM=${BUFFER_MEM:-33554432}
BATCH_SIZE=${BATCH_SIZE:-65536}
LINGER_MS=${LINGER_MS:-100}
ACKS=${ACKS:-1}
FETCH_SIZE=${FETCH_SIZE:-1048576}
ENCRYPTION=${ENCRYPTION:-false}
PAYLOAD_NUM=${PAYLOAD_NUM:-10000}
NUM_NETWORK_THREADS=${NUM_NETWORK_THREADS:-0}
SERVER_PROTECTION=${SERVER_PROTECTION:-true}
NUM_REPLICA_FETCHERS=${NUM_REPLICA_FETCHERS:-1}
REPLICA_FETCH_MAX_BYTES=${REPLICA_FETCH_MAX_BYTES:-1048576}
REPLICA_SOCKET_RECEIVE_BUFFER_BYTES=${REPLICA_SOCKET_RECEIVE_BUFFER_BYTES:-65536}
# Set PARTITIONS/PRODUCERS/CONSUMERS to 0 if value is not specified by user, will update this value at runtime in run_test.sh if value is 0
PARTITIONS=${PARTITIONS:-0}
PRODUCERS=${PRODUCERS:-0}
CONSUMERS=${CONSUMERS:-0}
# numactl setting
# default no numa control enable
SERVER_NUMACTL_OPTIONS=${SERVER_NUMACTL_OPTIONS:-""}
CONSUMER_NUMACTL_OPTIONS=${CONSUMER_NUMACTL_OPTIONS:-""}
PRODUCER_NUMACTL_OPTIONS=${PRODUCER_NUMACTL_OPTIONS:-""}
# JVM Setting
KAFKA_HEAP_OPTS=${KAFKA_HEAP_OPTS:-""}  #eg: -Xmx4G_-Xms4G  ,use _ connect each option
#support attach multiple disks
ENABLE_MUL_DISK=${ENABLE_MUL_DISK:-false} #Cloud platform only.Also set <CSP>_DISK_SPEC_1_DISK_COUNT equal to MOUNT_DISK_COUNT.
MOUNT_DISK_COUNT=${MOUNT_DISK_COUNT:-1} #mount dir like /mnt/disk1,/mnt/disk{N}, N is the number of disks
SERVER_CORE_NEEDED_FACTOR=${SERVER_CORE_NEEDED_FACTOR:-1}


# Logs Setting
DIR="$( cd "$( dirname "$0" )" &> /dev/null && pwd )"
. "$DIR/../../script/overwrite.sh"

if [[ $DEFAULT_OR_GATED == "gated" ]]; then
    PARTITIONS=1
    PRODUCERS=1
    CONSUMERS=1
    BROKER_SERVER_NUM=1
    PRODUCER_NODE_NUM=1
    CONSUMER_NODE_NUM=1
    TOPIC_NUM=1
fi

# Set SERVER_CORE_NEEDED_FACTOR, SERVER_CORE_NEEDED_FACTOR is only valid for single node scenario
if [[ ${TESTCASE} =~ _1n ]]; then
    PRODUCER_NODE_NUM=1
    CONSUMER_NODE_NUM=1
    BROKER_SERVER_NUM=1
    TOPIC_NUM=1
    if [[ `awk -v factor=$SERVER_CORE_NEEDED_FACTOR 'BEGIN {print factor <= 0 || factor >= 1 ? "1" : "0" }'` != 0 ]]; then
        SERVER_CORE_NEEDED_FACTOR=0.1
    fi
else
    SERVER_CORE_NEEDED_FACTOR=1
fi


# Workload Setting
WORKLOAD_PARAMS=(
    REPLICATION_FACTOR PARTITIONS NUM_RECORDS THROUGHPUT RECORD_SIZE COMPRESSION_TYPE
    MESSAGES PRODUCERS CONSUMERS CONSUMER_TIMEOUT BUFFER_MEM BATCH_SIZE LINGER_MS
    SERVER_CORE_NEEDED_FACTOR BROKER_SERVER_NUM ACKS FETCH_SIZE ENCRYPTION PAYLOAD_NUM
    NUM_NETWORK_THREADS SERVER_PROTECTION SERVER_NUMACTL_OPTIONS CONSUMER_NUMACTL_OPTIONS KAFKA_HEAP_OPTS
    PRODUCER_NUMACTL_OPTIONS NUM_REPLICA_FETCHERS REPLICA_FETCH_MAX_BYTES REPLICA_SOCKET_RECEIVE_BUFFER_BYTES
    ENABLE_MUL_DISK MOUNT_DISK_COUNT PRODUCER_NODE_NUM TOPIC_NUM CONSUMER_NODE_NUM
)

# Docker Setting
DOCKER_IMAGE=""
DOCKER_OPTIONS=""

# Kubernetes Setting
RECONFIG_OPTIONS="-DREPLICATION_FACTOR=${REPLICATION_FACTOR} -DPARTITIONS=${PARTITIONS} -DKAFKA_BENCHMARK_TOPIC=${KAFKA_BENCHMARK_TOPIC} \
    -DMESSAGES=${MESSAGES} -DNUM_RECORDS=${NUM_RECORDS} -DTHROUGHPUT=${THROUGHPUT} -DRECORD_SIZE=${RECORD_SIZE} -DCOMPRESSION_TYPE=${COMPRESSION_TYPE} \
    -DPRODUCERS=${PRODUCERS} -DCONSUMERS=${CONSUMERS} -DCONSUMER_TIMEOUT=${CONSUMER_TIMEOUT} -DBUFFER_MEM=${BUFFER_MEM} -DBATCH_SIZE=${BATCH_SIZE} -DLINGER_MS=${LINGER_MS} \
    -DSERVER_CORE_NEEDED_FACTOR=${SERVER_CORE_NEEDED_FACTOR} -DBROKER_SERVER_NUM=${BROKER_SERVER_NUM} -DACKS=${ACKS} -DFETCH_SIZE=${FETCH_SIZE} -DENCRYPTION=${ENCRYPTION} -DPAYLOAD_NUM=${PAYLOAD_NUM} \
    -DNUM_NETWORK_THREADS=${NUM_NETWORK_THREADS} -DSERVER_PROTECTION=${SERVER_PROTECTION} -DSERVER_NUMACTL_OPTIONS=${SERVER_NUMACTL_OPTIONS} -DCONSUMER_NUMACTL_OPTIONS=${CONSUMER_NUMACTL_OPTIONS} \
    -DPRODUCER_NUMACTL_OPTIONS=${PRODUCER_NUMACTL_OPTIONS} -DNUM_REPLICA_FETCHERS=${NUM_REPLICA_FETCHERS} -DREPLICA_FETCH_MAX_BYTES=${REPLICA_FETCH_MAX_BYTES} \
    -DREPLICA_SOCKET_RECEIVE_BUFFER_BYTES=${REPLICA_SOCKET_RECEIVE_BUFFER_BYTES} -DKAFKA_HEAP_OPTS=${KAFKA_HEAP_OPTS} \
    -DENABLE_MUL_DISK=${ENABLE_MUL_DISK} -DMOUNT_DISK_COUNT=${MOUNT_DISK_COUNT} -DPRODUCER_NODE_NUM=${PRODUCER_NODE_NUM} \
    -DTOPIC_NUM=${TOPIC_NUM} -DCONSUMER_NODE_NUM=${CONSUMER_NODE_NUM}"

# Used for log collection
JOB_FILTER="job-name=benchmark"

# kpi args
p95_latency_sla=${p95_latency_sla:-20}
p99_latency_sla=${p99_latency_sla:-40}
SCRIPT_ARGS="${p95_latency_sla} ${p99_latency_sla}"

EVENT_TRACE_PARAMS="roi,begin region of interest,end region of interest"

. "$DIR/../../script/validate.sh"
