generate-cosine-data-multi-entity.py

'''

Python script for ingesting sample data into OpenSearch index

'''

#{"index": {"_index":"host-cloudwatch","_id":"1177"}}
#{"@timestamp":"2017-03-23T13:00:00","cpu":20.3, "memory":13,"host":"host1", "service": "service1"}

import numpy as np
from scipy.stats import uniform
import datetime
import time
import random
from random import Random
from retry import retry
import urllib3
import concurrent.futures
import argparse

from opensearchpy import OpenSearch, RequestsHttpConnection
from opensearchpy import helpers

# https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings
urllib3.disable_warnings()

parser = argparse.ArgumentParser()

parser.add_argument("-ep", "--endpoint", help="cluster endpoint", required=True)
parser.add_argument("-i", "--index-name", help=" ",required=True)
parser.add_argument("-shards", "--shards", type=int, help="The number of shards for the given index", required=True)
parser.add_argument("-t", "--threads", type=int, help="The number of threads to be used for data ingestion, make sure given machine has enough", required=True)
parser.add_argument("-bulk", "--bulk-size", type=int, default=3000, help="Number of documents per bulk request, default to 3000", )
parser.add_argument("-ingest", "--ingestion-frequency", type=int, default=600, help="how often each respective document is indexed, for example the default is 600 seconds which equates to every 10 minutes")
parser.add_argument("-p", "--points", type=int, default=1008, help="total number of points ingested, for example with 1008 points and a frequency of 600s, there will be 7 days of data")

parser.add_argument('--security', action='store_true')
parser.add_argument('--no-security', dest='security', action='store_false')
parser.set_defaults(security=False)

parser.add_argument("-nh", "--number-of-host", type=int, default=1000, help="number of 'host' entities, deafult is set to 1000, there will be two keyword categories in this index (must be at least 1)")
parser.add_argument("-np", "--number-of-process", type=int, default=1000, help="number of 'process' entities, deafult is set to 1000, there will be two keyword categories in this index (must be at least 1)" )
parser.add_argument("-hd", "--number-of-historical-days", type=int, default=2, help="number of day of historical data to ingest, defaults to 2")
parser.add_argument("-u", "--username", type=str, default="admin", help="username for authentication if security is true")
parser.add_argument("-pass", "--password", type=str, default="admin", help="password for authentication if security is true")

args = parser.parse_args()

URL = args.endpoint
SECURITY = args.security
INDEX_NAME = args.index_name
SHARD_NUMBER = args.shards
THREADS = args.threads

#deafult numbers of 1000 host and 1000 process mean a total of 1 million entities
HOST_NUMBER = args.number_of_host
PROCESS_NUMBER = args.number_of_process

#default of 1008 points with ingestion frequency set to 600 means there will basically be 1008 intervals = 7 days * 144 intervals/day
POINTS = args.points 
INGESTION_FREQUENCY = args.ingestion_frequency
BULK_SIZE = args.bulk_size
USERNAME = args.username
PASSWORD = args.password
NUMBER_OF_HISTORICAL_DAYS = args.number_of_historical_days


index_name = "_index"
timestamp_name = "@timestamp"
cpu_name = "cpuTime"
mem_name = "jvmGcTime"
host_name = "host"
host_prefix = "host"
process_name = "process"
process_prefix = "process"
client = []

'''
    Generate index INDEX_NAME
'''
def create_index(os, INDEX_NAME, shard_number):
    # First, delete the index if it exists
    print("Deleting index if it exists...")
    os.indices.delete(index=INDEX_NAME, ignore=[400, 404])

    # Next, create the index
    print("Creating index \"{}\"...".format(INDEX_NAME))
    request_body = {
        "settings":{
          "number_of_shards":shard_number,
          "number_of_replicas": 0, # increase this number after indexing
          "translog.durability":"async", # default: request
          "refresh_interval":-1, # default: 1, remember to change this after finishing indexing process or just _refresh once at least if index wont be changed again
       },
       "mappings":{
          "properties":{
             "@timestamp":{
                "type":"date"
             },
             "cpuTime":{
                "type":"double"
             },
             "jvmGcTime":{
                "type":"double"
             },
             "host":{
                "type":"keyword"
             },
             "process":{
                "type":"keyword"
             }
          }
       }
    }
    os.indices.create(index=INDEX_NAME, body=request_body)

'''
    Posts a document(s) to the index
'''
@retry(delay=1, backoff=2)
def post_log(bulk_payload, thread_index):
    global client
    helpers.bulk(client[thread_index], bulk_payload)

def generate_val(amp, phase, base_dimension, index, period, noise, noiseprg):
    data = np.empty(base_dimension, dtype=float)
    for j in range(0, base_dimension):
        # cos is [-1, 1], + 1 make it non-negative
        data[j] = amp[j] * (np.cos(2 * np.pi * (index + phase[j]) / period) + 1) + noise * noiseprg.random()
        if (noiseprg.random() < 0.01 and noiseprg.random() < 0.3):
            factor = 5 * (1 + noiseprg.random())
            change = factor * noise if noiseprg.random() < 0.5 else -factor * noise
            if data[j] + change >= 0:
                data[j] += change
    return data

'''
    Posts all documents to index in stream
'''
def post_log_stream(index_value, time_intervals, sample_per_interval, max_number, min_number, host_number, service_number, batch_size, thread_index, cosine_params):
    # For each file, post all the docs
    print("Posting logs...")
    bulk_payload = list()
    # give some data in the history for cold start
    dtFormat = "%Y-%m-%dT%H:%M:%S"
    startTs = datetime.datetime.utcnow() - datetime.timedelta(days=NUMBER_OF_HISTORICAL_DAYS)
    count = 0
    totalCount = 0
    lastTotalCount = 0

    keep_loop = True
    j = (int)(min_number / service_number)
    index = j * service_number - 1

    retries = 0
    while keep_loop and retries < 10 and j < host_number:
        try:
            while keep_loop and j < host_number:
                host_str = host_prefix + str(j)
                for l in range(service_number):
                    process_str = process_prefix + str(l)
                    index += 1
                    # index can be [min_number, max_number]
                    if index < min_number:
                        continue
                    if index > max_number:
                        keep_loop = False
                        break
                    nextTs = startTs
                    prb = Random()
                    prb.seed(random.randint(0, 100000000))
                    cosine_p = cosine_params[index]
                    data_index = 0
                    for i in range(0, time_intervals):
                        ts = nextTs.strftime(dtFormat)
                        for k in range(0, sample_per_interval):
                            data = generate_val(cosine_p[1], cosine_p[0], 2, data_index,
                                                50, 5, prb)
                            bulk_payload.append(
                                {
                                    index_name: index_value,
                                    "_source":
                                        {
                                            timestamp_name: ts,
                                            cpu_name: data[0],
                                            mem_name: data[1],
                                            host_name: host_str,
                                            process_name: process_str
                                        }
                                }
                            )
                            count += 1
                            data_index += 1
                            if count >= batch_size:
                                post_log(bulk_payload, thread_index)
                                bulk_payload = list()  # reset list
                                totalCount += count
                                count = 0
                        # increment by ingestion_frequency (in seconds) after looping through each host multiple samples
                        nextTs += datetime.timedelta(seconds=INGESTION_FREQUENCY)
                        if totalCount - lastTotalCount > 1000000:
                            # report progress every 1 million inserts
                            print("totalCount {} thread_index {}".format(totalCount,
                                                                        thread_index))
                            lastTotalCount = totalCount
                j += 1

            if len(bulk_payload) > 0:
                post_log(bulk_payload, thread_index)
                bulk_payload = list()
        except Error as err:
            print("error: {0}".format(err))
            retries += 1
            client[thread_index] = create_client(SECURITY, URL)

def split(a, n):
    k, m = divmod(len(a), n)
    return (a[i*k+min(i, m):(i+1)*k+min(i+1, m)] for i in range(n))

# create an list of array of size total_entities, the inner array has 2 subarrays: phase, amp
def create_cosine(total_entities, base_dimension, period, amplitude):
    cosine_param = np.empty(total_entities, dtype=object)
    for i in range(0, total_entities):
        phase = np.empty(base_dimension, dtype=float)
        amp = np.empty(base_dimension, dtype=float)
        for j in range(0, base_dimension):
            phase[j] = random.randint(0, period)
            amp[j] = (1 + 0.2 * random.random()) * amplitude
        cosine_param[i] = np.array([phase, amp])
    return cosine_param

'''
Create OpenSearch client
'''
def create_client(security, URL):
    if security and URL.strip() == 'localhost':
        return OpenSearch(
            hosts=[URL],
            use_ssl=True,
            verify_certs=False,
            http_auth=(USERNAME, PASSWORD),
            scheme="https",
            connection_class=RequestsHttpConnection
         )
    elif security:
        return OpenSearch(
                hosts=[{'host': URL, 'port': 443}],
                use_ssl=True,
                verify_certs=False,
                http_auth=(USERNAME, PASSWORD),
                scheme="https",
                port=443,
                connection_class=RequestsHttpConnection
        )
    elif URL.strip() == 'localhost':
        return OpenSearch(
            hosts=[{'host': URL, 'port': 9200}],
            use_ssl=False,
            verify_certs=False,
            connection_class=RequestsHttpConnection
        )
    else:
        return OpenSearch(
            hosts=[{'host': URL, 'port': 80}],
            use_ssl=False,
            verify_certs=False,
            connection_class=RequestsHttpConnection
        )

'''
    Main entry method for script
'''
def main():
    global client
    for i in range(0, THREADS):
        client.append(create_client(SECURITY, URL))

    create_index(client[0], INDEX_NAME, SHARD_NUMBER)

    total_entities = HOST_NUMBER * PROCESS_NUMBER
    # https://tinyurl.com/yeter98e
    # workload is a list of ranges like [range(0, 10000), range(10000, 20000)]
    workload = list(split(range(total_entities), THREADS))
    futures = []

    # we we have both cpuTime and jvmGcTime field, so 2 features
    cosine_params = create_cosine(total_entities, 2, 50, 100)
    start = time.monotonic()
    with concurrent.futures.ProcessPoolExecutor(max_workers=THREADS) as executor:
        futures = []
        for i in range(len(workload)):
            # Using 1 sample per interval to reason about the result easier.
            doc_per_interval = 1
            futures.append(executor.submit(post_log_stream, INDEX_NAME, POINTS, doc_per_interval, workload[i][-1], workload[i][0], HOST_NUMBER, PROCESS_NUMBER, BULK_SIZE, i, cosine_params))
        _ = concurrent.futures.as_completed(futures)

    print('Concurrent took: %.2f minutes.' % ((time.monotonic() - start)/60))

if __name__ == "__main__":
    main()