extra/docker/elk/tools/DMaaPServiceMocker/ds_mocker.py

#!/usr/bin/env python3
import os
import json
import copy
import random
import requests
import uuid
import time
from datetime import datetime

def luck(n=2):
    """ gives 1 chance out of n (default: 2) to return True """
    assert n > 1
    return bool(random.randint(0, n-1))
def now_dmaap_timestamp():
    return str(datetime.now().timestamp()).replace(".","")[:13]
def now_notification_time():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f+00:00")

CONTROL_LOOP_NAMES = [
    'CL-vCPE-d925ed73',
    'CL-vCPE-37b1c91e',
    'CL-vCPE-c2597657',
    'CL-vCPE-a11318ba',
    'CL-vCPE-5321c558',
]

TEMPLATES = {
    'event_abated'                   :'event_abated.json',
    'event_onset'                    :'event_onset.json',
    'notification_active'            :'notification_active.json',
    'notification_final_failed'      :'notification_final_failed.json',
    'notification_final_open'        :'notification_final_open.json',
    'notification_final_success'     :'notification_final_success.json',
    'notification_operation_failure' :'notification_operation_failure.json',
    'notification_operation'         :'notification_operation.json',
    'notification_operation_success' :'notification_operation_success.json',
    'notification_rejected_disabled' :'notification_rejected_disabled.json',
    'notification_rejected_missing'  :'notification_rejected_missing.json',
}

ERROR_MESSAGES = [
    ('APPC', 'APPC1 : timeout on restart','RESTART'),
    ('APPC', 'APPC2 : cannot restart','RESTART'),
    ('SO', 'SO1 : scale up failed', 'SCALEUP'),
    ('SO', 'SO2 : scale down failed', 'SCALEDOWN'),
]

for key in TEMPLATES:
    with open(TEMPLATES[key]) as f:
        content = f.read()
    TEMPLATES[key] = json.loads(content)


class DMaaPMessage(dict):

    dmaap_host_url = "http://dmaap.host.url:9200/"
    dmaap_username = None
    dmaap_password = None

    @classmethod
    def from_template(cls, tmpl, **kwargs):
        obj = cls()
        obj.update(copy.deepcopy(TEMPLATES[tmpl]))
        for keys,value in kwargs.items():
            current_node = obj
            keys = keys.split(".")
            key = keys[0]
            for i in range(len(keys) - 1):
                current_node = current_node[keys[i]]
                key = keys[i]
            current_node[key] = value
        return obj

    def publish(self, topic):
        url = "%s/events/%s" % (self.dmaap_host_url, topic)
        auth = None
        if self.dmaap_username and self.dmaap_password:
            auth = (self.dmaap_username, self.dmaap_password)
        response = requests.post(url, data=json.dumps(self), auth=auth)
        return response.status_code

class Event(DMaaPMessage):

    topic = "DCAE-CL-EVENT"

    @staticmethod
    def abated(**kwargs):
        return Event.from_template('event_abated', **kwargs)

    @staticmethod
    def onset(**kwargs):
        return Event.from_template('event_onset', **kwargs)

    def publish(self):
        return super().publish(self.topic)


class Notification(DMaaPMessage):

    topic = "POLICY-CL-MGT"

    @classmethod
    def from_template(cls, tmpl, **kwargs):
        kwargs['notificationTime'] = now_notification_time()
        return super().from_template(tmpl, **kwargs)

    @staticmethod
    def active(**kwargs):
        return Notification.from_template('notification_active', **kwargs)

    @staticmethod
    def final(**kwargs):
        class FinalNotification(Notification):
            @staticmethod
            def success(**kwargs):
                return FinalNotification.from_template('notification_final_success', **kwargs)
            @staticmethod
            def failed(**kwargs):
                msg = FinalNotification.from_template('notification_final_failed', **kwargs)
                error = ERROR_MESSAGES[random.randint(0, len(ERROR_MESSAGES) - 1)]
                h = msg['history'][-1]
                h['actor'],h['message'],h['operation'] = error[0],error[1],error[2]
                return msg
            @staticmethod
            def open(**kwargs):
                return FinalNotification.from_template('notification_final_open', **kwargs)
        return FinalNotification

    @staticmethod
    def operation(**kwargs):
        class OperationNotification(Notification):
            @staticmethod
            def success(**kwargs):
                return OperationNotification.from_template('notification_operation_success', **kwargs)
            @staticmethod
            def failure(**kwargs):
                return OperationNotification.from_template('notification_operation_failure', **kwargs)
        return OperationNotification.from_template('notification_operation', **kwargs)

    @staticmethod
    def rejected(**kwargs):
        class RejectedNotification(Notification):
            @staticmethod
            def disabled(**kwargs):
                return RejectedNotification.from_template('notification_rejected_disabled', **kwargs)
            @staticmethod
            def missing_fields(**kwargs):
                return RejectedNotification.from_template('notification_rejected_missing', **kwargs)

        return RejectedNotification

    def publish(self):
        return super().publish(self.topic)



class CLStatus(object):

    def __init__(self, dmaap_url=None,
                 missing=None, disabled=None, op_failure=None):
        self._stopped = False
        def maybe(thing, ):
            if thing is None:
                thing = not luck(10)
            return thing
        self._missing = maybe(missing)
        self._disabled = maybe(disabled)
        self._op_failure = maybe(op_failure)
        self._config = dict(
            requestID=str(uuid.uuid4()),
            closedLoopControlName=CONTROL_LOOP_NAMES[random.randint(0, len(CONTROL_LOOP_NAMES) - 1)]
        )

    def __iter__(self):
        return next(self)

    def __next__(self):
        if self._stopped:
            raise StopIteration()
        config = self._config
        config.update(dict(closedLoopAlarmStart=now_dmaap_timestamp()))
        yield Event.onset(**config)
        if self._missing:
            self._stopped = True
            yield Notification.rejected().missing_fields(**config)
            raise StopIteration()
        elif self._disabled:
            self._stopped = True
            yield Notification.rejected().disabled(**config)
            raise StopIteration()

        yield Notification.active(**config)
        yield Notification.operation(**config)

        config['closedLoopAlarmEnd'] = now_dmaap_timestamp()
        if self._op_failure:
            yield Notification.operation().failure(**config)
            self._stopped = True
            yield Notification.final().failed(**config)
        else:
            yield Notification.operation().success(**config)
            yield Event.abated(**config)
            self._stopped = True
            yield Notification.final().success(**config)
        raise StopIteration()

def print_usage():
    print("""
    ./ds_mocker.py <DMAAP_URL> <EVENT_TOPIC> [NOTIFICATION_TOPIC [REQUEST_TOPIC]]
    """)
    exit()

def push(test_datas):
    for current_i, status in enumerate(test_datas):
        time.sleep(random.randint(0,3))
        for s in status:
            # print(s)
            status_code = s.publish()
            if status_code != 200:
                print("Error when publishing : status_code={}".format(status_code))
                exit(1)
            time.sleep(random.randint(0,3))
        print("%03d,missing:%5s,disabled:%5s,op_failure:%5s - %s" % (current_i, status._missing, status._disabled, status._op_failure, status._config))



def generate_dataset_1():
    test_datas = [CLStatus(missing=False, disabled=False, op_failure=False) for i in range(300)]  \
             + [CLStatus(missing=True, disabled=False, op_failure=False) for i in range(5)]  \
             + [CLStatus(missing=False, disabled=True, op_failure=False) for i in range(6)]  \
             + [CLStatus(missing=False, disabled=False, op_failure=True) for i in range(12)]
    random.shuffle(test_datas)
    return test_datas

def generate_error_dataset_1():
    test_datas = [CLStatus(missing=False, disabled=False, op_failure=True) for i in range(60)]
    random.shuffle(test_datas)
    return test_datas


DATASETS = {
    'dataset_1': generate_dataset_1,
    'op_failure_1': generate_error_dataset_1,
}

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 3:
        print_usage()

    DMaaPMessage.dmaap_host_url = sys.argv[1]
    Event.topic = sys.argv[2]
    Notification.topic = len(sys.argv) > 3 and sys.argv[3] or sys.argv[2]
    # Request.topic = len(sys.argv) > 4 or Notification.topic
    #push(DATASETS['op_failure_1']())
    push(DATASETS['dataset_1']())