WFP-01-02-01-US-03 Data pipeline definition

As a data pipeline developer, I want to define a data pipeline trigger for the application deployed in WFP-01-02-01-US-02, with the following configurable parameters:

• Data stream unique identifier
• Area of interest (AOI)
• Time range (start date / end date)
• Data polling period

Build the trigger

The Production Centre uses Docker images and deploys them in a Cloud based environment for large scale production campaigns.

This notebook will trigger the build of a Docker image containing the trigger.

We will invoke components of the Ellip Platform Continuous Integration.

The input for triggering the build of Docker image containing the data transformation application uses the pom, an XML file that contains information about the data transformation application and configuration details used by Maven to build the Docker image.

The goal of this Jupyter Notebook is to build of the data transformation application Docker image.

• A few Python modules to do the job:

import os
import time

import jenkins

from nbconvert.preprocessors import ExecutePreprocessor, CellExecutionError
import nbformat as nbf

• Read the data transformation application information

%store -r

nb_config = os.path.join('..', 'configuration.ipynb')

nb = nbf.read(nb_config, 4)

exec(nb['cells'][1]['source']) in globals(), locals()

trigger_queue = dict([('artifact_id', trigger_queue_artifact_id),
                      ('repository', repository),
                      ('folder', folder),
                      ('community', community)])

trigger_pipe = dict([('artifact_id', trigger_pipe_artifact_id),
                      ('repository', repository),
                      ('folder', folder),
                      ('community', community)])

• Create two simple functions to submit the build request and to monitor it:

def submit_build_job(server, job, sleep_secs=3):

    last_build = server.get_job_info(job)['lastCompletedBuild']['number']

    server.build_job(job)

    while server.get_job_info(job)['lastBuild']['number'] == last_build:

        time.sleep(sleep_secs)

    return server.get_job_info(job)['lastBuild']['number']

def monitor_build_job(server, job, sleep_secs=15):

    last_build_number = server.get_job_info(job)['lastBuild']['number']

    while server.get_build_info(job, last_build_number)['building']:

        time.sleep(sleep_secs)

    if str(server.get_build_info(job, last_build_number)['result']) == 'SUCCESS':

        return True

    if str(server.get_build_info(job, last_build_number)['result']) == 'FAILURE':

        return False

• Establish a connection with the build component of the Ellip Platform Continuous Integration:

server = jenkins.Jenkins(url=jenkins_url,
                         username=jenkins_username,
                         password=jenkins_api_key)

• Submit the build request for the two trigger applications:

for trigger in [trigger_queue, trigger_pipe]:

    job = 'communities/%s/%s/%s/triggers/%s/docker' % (trigger['repository'],
                                                       trigger['community'],
                                                       trigger['folder'],
                                                       trigger['artifact_id'])

    print 'Submit build for job %s...' % job

    submit_build_job(server, job)

    print 'Build for job %s submitted...' % job

    if monitor_build_job(server, job):
        print 'Job build for %s completed' % trigger['artifact_id']
    else:
        print 'Job build for %s failed' % trigger['artifact_id']

Submit build for job communities/Gitlab Groups/ec-better/wfp/triggers/tg-wfp-01-02-01-queue/docker...
Build for job communities/Gitlab Groups/ec-better/wfp/triggers/tg-wfp-01-02-01-queue/docker submitted...
Job build for tg-wfp-01-02-01-queue completed
Submit build for job communities/Gitlab Groups/ec-better/wfp/triggers/tg-wfp-01-02-01-pipe/docker...
Build for job communities/Gitlab Groups/ec-better/wfp/triggers/tg-wfp-01-02-01-pipe/docker submitted...
Job build for tg-wfp-01-02-01-pipe completed