Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 35 additions & 0 deletions .github/resources/github_workflow.properties
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
# The script that abides by the multi-language protocol. This script will
# be executed by the MultiLangDaemon, which will communicate with this script
# over STDIN and STDOUT according to the multi-language protocol.
executableName = sample_kclpy_app.py

# The name of an Amazon Kinesis stream to process.
# Important: streamArn takes precedence over streamName if both are set
streamName = STREAM_NAME_PLACEHOLDER

# Used by the KCL as the name of this application. Will be used as the name
# of an Amazon DynamoDB table which will store the lease and checkpoint
# information for workers with this application name
applicationName = APP_NAME_PLACEHOLDER

# Users can change the credentials provider the KCL will use to retrieve credentials.
# Expected key name (case-sensitive):
# AwsCredentialsProvider / AwsCredentialsProviderDynamoDB / AwsCredentialsProviderCloudWatch
# The DefaultCredentialsProvider checks several other providers, which is
# described here:
# https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html
AwsCredentialsProvider = DefaultCredentialsProvider

# Appended to the user agent of the KCL. Does not impact the functionality of the
# KCL in any other way.
processingLanguage = python/3.8

# Valid options at TRIM_HORIZON or LATEST.
# See http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html#API_GetShardIterator_RequestSyntax
initialPositionInStream = TRIM_HORIZON

# The KCL defaults to us-east-1
regionName = us-east-1

# Idle time between record reads in milliseconds.
idleTimeBetweenReadsInMillis = 250
31 changes: 31 additions & 0 deletions .github/scripts/clean_up_stream_table.sh
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
#!/bin/bash

# Delete stream
if aws kinesis describe-stream --stream-name $STREAM_NAME &>/dev/null; then
echo "Deleting stream $STREAM_NAME"
for i in {1..10}; do
aws kinesis delete-stream --stream-name $STREAM_NAME && break ||
echo "Stream deletion failed, attempt $i/10. Retrying Stream deletion in $((i * 3))s" && sleep $((i * 3))
done
else
echo "Stream $STREAM_NAME does not exist and does not need to be cleaned up"
fi

# Delete table
delete_table() {
table_name=$1
if aws dynamodb describe-table --table-name $table_name &>/dev/null; then
echo "Deleting table $table_name"
for i in {1..10}; do
aws dynamodb delete-table --table-name $table_name && break ||
echo "Table deletion failed, attempt $i/10. Retrying DynamoDB Table deletion in $((i * 3))s" && sleep $((i * 3))
done
else
echo "Table $table_name does not exist and does not need to be cleaned up"
fi
}

# Delete all tables
for SUFFIX in "" "-CoordinatorState" "-WorkerMetricStats"; do
delete_table "$APP_NAME$SUFFIX"
done
12 changes: 12 additions & 0 deletions .github/scripts/create_stream.sh
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
#!/bin/bash
set -e

for i in {1..10}; do
if aws kinesis create-stream --stream-name $STREAM_NAME --shard-count 1; then
break
else
echo "Stream creation failed, attempt $i/10. Waiting $((i * 3)) seconds..."
sleep $((i * 3))
fi
done
aws kinesis wait stream-exists --stream-name $STREAM_NAME
23 changes: 23 additions & 0 deletions .github/scripts/manipulate_properties.sh
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
#!/bin/bash
set -e

# Manipulate github_workflow.properties file that the KCL application pulls properties from (ex: streamName, applicationName)
# Depending on the OS, different properties need to be changed
if [[ "$RUNNER_OS" == "macOS" ]]; then
sed -i "" "s/STREAM_NAME_PLACEHOLDER/$STREAM_NAME/g" .github/resources/github_workflow.properties
sed -i "" "s/APP_NAME_PLACEHOLDER/$APP_NAME/g" .github/resources/github_workflow.properties
elif [[ "$RUNNER_OS" == "Linux" || "$RUNNER_OS" == "Windows" ]]; then
sed -i "s/STREAM_NAME_PLACEHOLDER/$STREAM_NAME/g" .github/resources/github_workflow.properties
sed -i "s/APP_NAME_PLACEHOLDER/$APP_NAME/g" .github/resources/github_workflow.properties

if [[ "$RUNNER_OS" == "Windows" ]]; then
echo '@echo off' > samples/run_script.bat
echo 'python %~dp0\sample_kclpy_app.py %*' >> samples/run_script.bat
sed -i 's/executableName = sample_kclpy_app.py/executableName = samples\/run_script.bat/' .github/resources/github_workflow.properties
fi
else
echo "Unknown OS: $RUNNER_OS"
exit 1
fi

cat .github/resources/github_workflow.properties
15 changes: 15 additions & 0 deletions .github/scripts/put_words_to_stream.sh
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
#!/bin/bash
set -e

sample_kinesis_wordputter.py --stream $STREAM_NAME -w cat -w dog -w bird -w lobster -w octopus

# Get records from stream to verify they exist before continuing
SHARD_ITERATOR=$(aws kinesis get-shard-iterator --stream-name $STREAM_NAME --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --query 'ShardIterator' --output text)
INITIAL_RECORDS=$(aws kinesis get-records --shard-iterator $SHARD_ITERATOR)
RECORD_COUNT_BEFORE=$(echo $INITIAL_RECORDS | jq '.Records | length')

if [ "$RECORD_COUNT_BEFORE" -eq 0 ]; then
echo "No records found in stream. Test cannot proceed."
exit 1
fi
echo "Found $RECORD_COUNT_BEFORE records in stream before KCL start"
21 changes: 21 additions & 0 deletions .github/scripts/start_kcl.sh
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
#!/bin/bash
set -e
set -o pipefail

chmod +x .github/resources/github_workflow.properties
chmod +x samples/sample_kclpy_app.py

if [[ "$RUNNER_OS" == "macOS" ]]; then
brew install coreutils
KCL_COMMAND=$(amazon_kclpy_helper.py --print_command --java $(which java) --properties .github/resources/github_workflow.properties)
gtimeout $RUN_TIME_SECONDS $KCL_COMMAND 2>&1 | tee kcl_output.log || [ $? -eq 124 ]
elif [[ "$RUNNER_OS" == "Linux" || "$RUNNER_OS" == "Windows" ]]; then
KCL_COMMAND=$(amazon_kclpy_helper.py --print_command --java $(which java) --properties .github/resources/github_workflow.properties)
timeout $RUN_TIME_SECONDS $KCL_COMMAND 2>&1 | tee kcl_output.log || [ $? -eq 124 ]
else
echo "Unknown OS: $RUNNER_OS"
exit 1
fi

echo "==========ERROR LOGS=========="
grep -i error kcl_output.log || echo "No errors found in logs"
20 changes: 20 additions & 0 deletions .github/scripts/verify_kcl.sh
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
#!/bin/bash
set -e

NUM_LEASES_FOUND=$(aws dynamodb scan --table-name $APP_NAME --select "COUNT" --query "Count" --output text || echo "0")
NUM_CHECKPOINTS_FOUND=$(aws dynamodb scan --table-name $APP_NAME --select "COUNT" --filter-expression "attribute_exists(checkpoint) AND checkpoint <> :trim_horizon" --expression-attribute-values '{":trim_horizon": {"S": "TRIM_HORIZON"}}' --query "Count" --output text || echo "0")

echo "Found $NUM_LEASES_FOUND leases and $NUM_CHECKPOINTS_FOUND non-TRIM-HORIZON checkpoint in DynamoDB"

echo "Printing checkpoint values"
aws dynamodb scan --table-name $APP_NAME --projection-expression "leaseKey,checkpoint" --output json

if [ "$NUM_LEASES_FOUND" -gt 0 ] && [ "$NUM_CHECKPOINTS_FOUND" -gt 0 ]; then
echo "Test passed: Found both leases and non-TRIM_HORIZON checkpoints in DDB (KCL is fully functional)"
exit 0
else
echo "Test failed: KCL not fully functional"
echo "Lease(s) found: $NUM_LEASES_FOUND"
echo "non-TRIM_HORIZON checkpoint(s) found: $NUM_CHECKPOINTS_FOUND"
exit 1
fi
112 changes: 0 additions & 112 deletions .github/workflows/privileged-run.yml

This file was deleted.

Loading
Loading