Merge pull request #147 from arjunshenoymec/master

adding a new env variable that provides an option between sequential …

Author: Anand Sanmukhani

app.py

@@ -3,7 +3,9 @@
 import os
 import logging
 from datetime import datetime
-from multiprocessing import Process, Queue
+from multiprocessing import Pool, Process, Queue
+from multiprocessing import cpu_count
+from functools import partial
 from queue import Empty as EmptyQueueException
 import tornado.ioloop
 import tornado.web
@@ -117,37 +119,50 @@ def make_app(data_queue):
         ]
     )
 
+def train_individual_model(predictor_model, initial_run):
+    metric_to_predict = predictor_model.metric
+    pc = PrometheusConnect(
+    url=Configuration.prometheus_url,
+    headers=Configuration.prom_connect_headers,
+    disable_ssl=True,
+    )
 
-def train_model(initial_run=False, data_queue=None):
-    """Train the machine learning model."""
-    for predictor_model in PREDICTOR_MODEL_LIST:
-        metric_to_predict = predictor_model.metric
-        data_start_time = datetime.now() - Configuration.metric_chunk_size
-        if initial_run:
-            data_start_time = (
-                datetime.now() - Configuration.rolling_training_window_size
-            )
-
-        # Download new metric data from prometheus
-        new_metric_data = pc.get_metric_range_data(
-            metric_name=metric_to_predict.metric_name,
-            label_config=metric_to_predict.label_config,
-            start_time=data_start_time,
-            end_time=datetime.now(),
-        )[0]
-
-        # Train the new model
-        start_time = datetime.now()
-        predictor_model.train(
-            new_metric_data, Configuration.retraining_interval_minutes
-        )
-        _LOGGER.info(
-            "Total Training time taken = %s, for metric: %s %s",
-            str(datetime.now() - start_time),
-            metric_to_predict.metric_name,
-            metric_to_predict.label_config,
+    data_start_time = datetime.now() - Configuration.metric_chunk_size
+    if initial_run:
+        data_start_time = (
+            datetime.now() - Configuration.rolling_training_window_size
         )
 
+    # Download new metric data from prometheus
+    new_metric_data = pc.get_metric_range_data(
+        metric_name=metric_to_predict.metric_name,
+        label_config=metric_to_predict.label_config,
+        start_time=data_start_time,
+        end_time=datetime.now(),
+    )[0]
+
+    # Train the new model
+    start_time = datetime.now()
+    predictor_model.train(
+            new_metric_data, Configuration.retraining_interval_minutes)
+
+    _LOGGER.info(
+        "Total Training time taken = %s, for metric: %s %s",
+        str(datetime.now() - start_time),
+        metric_to_predict.metric_name,
+        metric_to_predict.label_config,
+    )
+    return predictor_model
+
+def train_model(initial_run=False, data_queue=None):
+    """Train the machine learning model."""
+    global PREDICTOR_MODEL_LIST
+    parallelism = min(Configuration.parallelism, cpu_count())
+    _LOGGER.info(f"Training models using ProcessPool of size:{parallelism}")
+    training_partial = partial(train_individual_model, initial_run=initial_run)
+    with Pool(parallelism) as p:
+        result = p.map(training_partial, PREDICTOR_MODEL_LIST)
+    PREDICTOR_MODEL_LIST = result
     data_queue.put(PREDICTOR_MODEL_LIST)
 
 

configuration.py

@@ -57,3 +57,10 @@ class Configuration:
         "Metric data rolling training window size: %s", rolling_training_window_size
     )
     _LOGGER.info("Model retraining interval: %s minutes", retraining_interval_minutes)
+
+    # An option for Parallelism.
+    # An Integer specifying the number of metrics to be trained in parallel.
+    # Default: 1.
+    # Note: The upper limit to this will be decided by the number of CPU cores 
+    # available to the container.
+    parallelism = int(os.getenv("FLT_PARALLELISM", "1"))