Jenkins pipeline for producer

consumer-java/aws-kinesis-video-producer-sdk-canary-consumer/Makefile

@@ -0,0 +1,12 @@
+test: build
+
+install-system-dependencies:
+	@echo "Installing system dependencies"
+
+build: install-system-dependencies
+	@echo "Build java parser based consumer"
+	mvn package
+	# Create a temporary filename in /tmp directory
+	touch tmp_jar
+	# Create classpath string of dependencies from the local repository to a file
+	mvn -Dmdep.outputFile=tmp_jar dependency:build-classpath

producer-c/producer-cloudwatch-integ/CMake/Dependencies/libcppsdk-CMakeLists.txt

@@ -11,6 +11,7 @@ ExternalProject_Add(libawscpp-download
                 	  -DBUILD_DEPS=ON 
                 	  -DBUILD_SHARED_LIBS=OFF 
                 	  -DBUILD_ONLY=monitoring|logs
+                      -DCMAKE_INSTALL_PREFIX=${OPEN_SRC_INSTALL_PREFIX}
     BUILD_ALWAYS      TRUE
     TEST_COMMAND      ""
 )

producer-c/producer-cloudwatch-integ/CMake/Dependencies/libkvsProducerC-CMakeLists.txt

@@ -6,7 +6,7 @@ include(ExternalProject)
 
 ExternalProject_Add(libkvsProducerC-download
     GIT_REPOSITORY    https://github.com/awslabs/amazon-kinesis-video-streams-producer-c.git
-    GIT_TAG           origin/master
+    GIT_TAG           22edebfd87f5a38ab8af58da9a42f3d8dc7aebe7
     PREFIX            ${CMAKE_CURRENT_BINARY_DIR}/build
     CMAKE_ARGS        -DCMAKE_INSTALL_PREFIX=${OPEN_SRC_INSTALL_PREFIX} -DBUILD_COMMON_LWS=ON -DBUILD_COMMON_CURL=ON -DBUILD_DEPENDENCIES=TRUE -DOPEN_SRC_INSTALL_PREFIX=${OPEN_SRC_INSTALL_PREFIX}
     BUILD_ALWAYS      TRUE

producer-c/producer-cloudwatch-integ/canary/KvsProducerSampleCloudwatch.cpp

@@ -374,6 +374,6 @@ INT32 main(INT32 argc, CHAR* argv[])
         RESET_INSTRUMENTED_ALLOCATORS();
         DLOGI("CleanUp Done");
     }
-    DLOGD("Exiting application with status code: 0x%08x", retStatus);
+    DLOGI("Exiting application with status code: 0x%08x", retStatus);
     return STATUS_FAILED(retStatus) ? EXIT_FAILURE : EXIT_SUCCESS;
 }

producer-c/producer-cloudwatch-integ/jobs/canary_seed.groovy

@@ -0,0 +1,84 @@
+import java.lang.reflect.*;
+import jenkins.model.*;
+import org.jenkinsci.plugins.scriptsecurity.scripts.*;
+import org.jenkinsci.plugins.scriptsecurity.sandbox.whitelists.*;
+
+WORKSPACE="producer-c/producer-cloudwatch-integ"
+GROOVY_SCRIPT_DIR="$WORKSPACE/jobs"
+DAYS_TO_KEEP_LOGS=7
+NUMBER_OF_LOGS=5
+MAX_EXECUTION_PER_NODE=1
+NAMESPACE="producer"
+
+void approveSignatures(ArrayList<String> signatures) {
+    scriptApproval = ScriptApproval.get()
+    alreadyApproved = new HashSet<>(Arrays.asList(scriptApproval.getApprovedSignatures()))
+    signatures.each {
+      if (!alreadyApproved.contains(it)) {
+        scriptApproval.approveSignature(it)
+      }
+    }
+}
+
+approveSignatures([
+    "method hudson.model.ItemGroup getAllItems java.lang.Class",
+    "method hudson.model.Job getBuilds",
+    "method hudson.model.Job getLastBuild",
+    "method hudson.model.Job isBuilding",
+    "method hudson.model.Run getTimeInMillis",
+    "method hudson.model.Run isBuilding",
+    "method jenkins.model.Jenkins getItemByFullName java.lang.String",
+    "method jenkins.model.ParameterizedJobMixIn\$ParameterizedJob isDisabled",
+    "method jenkins.model.ParameterizedJobMixIn\$ParameterizedJob setDisabled boolean",
+    "method org.jenkinsci.plugins.workflow.job.WorkflowRun doKill",
+    "staticMethod jenkins.model.Jenkins getInstance",
+    "staticField java.lang.Long MAX_VALUE"
+])
+
+pipelineJob("${NAMESPACE}-orchestrator") {
+	description('Run producer and consumer')
+	logRotator {
+		daysToKeep(DAYS_TO_KEEP_LOGS)
+		numToKeep(NUMBER_OF_LOGS)
+	}
+    throttleConcurrentBuilds {
+        maxPerNode(MAX_EXECUTION_PER_NODE)
+    }
+
+	definition {
+        cps {
+        	script(readFileFromWorkspace("${GROOVY_SCRIPT_DIR}/orchestrator.groovy"))
+            sandbox()
+        }
+    }
+}
+
+pipelineJob("${NAMESPACE}-runner-0") {
+	description('Run producer and consumer')
+	logRotator {
+		daysToKeep(DAYS_TO_KEEP_LOGS)
+		numToKeep(NUMBER_OF_LOGS)
+	}
+	definition {
+        cps {
+        	script(readFileFromWorkspace("${GROOVY_SCRIPT_DIR}/runner.groovy"))
+            sandbox()
+        }
+    }
+}
+
+pipelineJob("${NAMESPACE}-runner-1") {
+    description('Run producer and consumer')
+    logRotator {
+        daysToKeep(DAYS_TO_KEEP_LOGS)
+        numToKeep(NUMBER_OF_LOGS)
+    }
+    definition {
+        cps {
+            script(readFileFromWorkspace("${GROOVY_SCRIPT_DIR}/runner.groovy"))
+            sandbox()
+        }
+    }
+}
+
+queue("${NAMESPACE}-orchestrator")

producer-c/producer-cloudwatch-integ/jobs/orchestrator.groovy

@@ -0,0 +1,160 @@
+import jenkins.model.*
+
+WORKSPACE_PRODUCER="producer-c/producer-cloudwatch-integ"
+WORKSPACE_CONSUMER="consumer-java/aws-kinesis-video-producer-sdk-canary-consumer"
+GIT_URL='https://github.com/aws-samples/amazon-kinesis-video-streams-demos.git'
+GIT_HASH='master'
+RUNNER_JOB_NAME_PREFIX = "producer-runner"
+
+// TODO: Set up configurability to run different parameter combinations
+// Run long run canary for 12 hours. New changes in the SDK will be pulled
+// in every 12 hours
+CANARY_DURATION_IN_SECONDS = 12 * 60 * 60
+COLD_STARTUP_DELAY_IN_SECONDS = 60 * 60
+MIN_RETRY_DELAY_IN_SECONDS = 60
+FRAGMENT_SIZE_IN_BYTES = 1048576
+AWS_DEFAULT_REGION = "us-west-2"
+
+COMMON_PARAMS = [
+    string(name: 'AWS_KVS_LOG_LEVEL', value: "2"),
+    string(name: 'GIT_URL', value: GIT_URL),
+    string(name: 'GIT_HASH', value: GIT_HASH),
+    string(name: 'MIN_RETRY_DELAY_IN_SECONDS', value: MIN_RETRY_DELAY_IN_SECONDS.toString()),
+]
+
+def getJobLastBuildTimestamp(job) {
+    def timestamp = 0
+    def lastBuild = job.getLastBuild()
+
+    if (lastBuild != null) {
+        timestamp = lastBuild.getTimeInMillis()
+    }
+
+    return timestamp
+}
+
+def cancelJob(jobName) {
+    def job = Jenkins.instance.getItemByFullName(jobName)
+
+    echo "Tear down ${jobName}"
+    job.setDisabled(true)
+    job.getBuilds()
+       .findAll({ build -> build.isBuilding() })
+       .each({ build -> 
+            echo "Kill $build"
+            build.doKill()
+        })
+}
+
+def findRunners() {
+    def filterClosure = { item -> item.getDisplayName().startsWith(RUNNER_JOB_NAME_PREFIX) }
+    return Jenkins.instance
+                    .getAllItems(Job.class)
+                    .findAll(filterClosure)
+}
+
+NEXT_AVAILABLE_RUNNER = null
+ACTIVE_RUNNERS = [] 
+
+pipeline {
+    agent {
+        label 'master'
+    }
+
+    options {
+        disableConcurrentBuilds()
+    }
+
+    stages {
+        stage('Checkout') {
+            steps {
+                checkout([$class: 'GitSCM', branches: [[name: GIT_HASH ]],
+                          userRemoteConfigs: [[url: GIT_URL]]])
+            }
+        }
+
+        stage('Update runners') {
+            stages {
+                stage("Find the next available runner and current active runners") {
+                    steps {
+                        script {
+                            def runners = findRunners()
+                            def nextRunner = null 
+                            def oldestTimestamp = Long.MAX_VALUE
+
+                            // find the least active runner
+                            runners.each {
+                                def timestamp = getJobLastBuildTimestamp(it)
+                                if ((it.isDisabled() || !it.isBuilding()) && timestamp < oldestTimestamp) {
+                                    nextRunner = it
+                                    oldestTimestamp = timestamp
+                                }
+                            }
+
+                            if (nextRunner == null) {
+                                error "There's no available runner"
+                            }
+
+                            NEXT_AVAILABLE_RUNNER = nextRunner.getDisplayName()
+                            echo "Found next available runner: ${NEXT_AVAILABLE_RUNNER}"
+
+                            ACTIVE_RUNNERS = runners.findAll({ item -> item != nextRunner && (!item.isDisabled() || item.isBuilding()) })
+                                                    .collect({ item -> item.getDisplayName() })
+                            echo "Found current active runners: ${ACTIVE_RUNNERS}"
+                        }
+                    }
+                }
+
+                stage("Spawn new runners") {
+                    steps {
+                        script {
+                            echo "New runner: ${NEXT_AVAILABLE_RUNNER}"
+                            Jenkins.instance.getItemByFullName(NEXT_AVAILABLE_RUNNER).setDisabled(false)
+
+                            def gitHash = sh(returnStdout: true, script: 'git rev-parse HEAD')
+                            COMMON_PARAMS << string(name: 'GIT_HASH', value: gitHash)
+                        }
+
+                        // TODO: Use matrix to provide configurability in parameters
+                        build(
+                            job: NEXT_AVAILABLE_RUNNER,
+                            parameters: COMMON_PARAMS + [
+                                string(name: 'CANARY_DURATION_IN_SECONDS', value: CANARY_DURATION_IN_SECONDS.toString()),
+                                string(name: 'PRODUCER_NODE_LABEL', value: "producer-uw2"),
+                                string(name: 'CONSUMER_NODE_LABEL', value: "consumer-uw2"),
+                                string(name: 'CANARY_TYPE', value: "Realtime"),
+                                string(name: 'RUNNER_LABEL', value: "Longrun"),
+                                string(name: 'FRAGMENT_SIZE_IN_BYTES', value: FRAGMENT_SIZE_IN_BYTES.toString()),
+                                string(name: 'AWS_DEFAULT_REGION', value: AWS_DEFAULT_REGION),
+                            ],
+                            wait: false
+                        )
+                    }
+                }
+
+                stage("Tear down old runners") {
+                    when {
+                        expression { return ACTIVE_RUNNERS.size() > 0 }
+                    }
+
+                    steps {
+                        script {
+                            try {
+                                sleep COLD_STARTUP_DELAY_IN_SECONDS
+                            } catch(err) {
+                                // rollback the newly spawned runner
+                                echo "Rolling back ${NEXT_AVAILABLE_RUNNER}"
+                                cancelJob(NEXT_AVAILABLE_RUNNER)
+                                throw err
+                            }
+
+                            for (def runner in ACTIVE_RUNNERS) {
+                                cancelJob(runner)
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+}