Add pl conf (#336)

Add pl conf --------- Signed-off-by: Nandita Koppisetty <[email protected]> Signed-off-by: s0nicboOm <[email protected]> Co-authored-by: Nandita Koppisetty <[email protected]>
numaproj · Dec 21, 2023 · b292553 · b292553
1 parent 28fa28f
commit b292553
Show file tree

Hide file tree

Showing 29 changed files with 1,708 additions and 1,247 deletions.
diff --git a/numalogic/backtest/_prom.py b/numalogic/backtest/_prom.py
@@ -37,7 +37,7 @@
 from numalogic.connectors.prometheus import PrometheusFetcher
 from numalogic.tools.data import StreamingDataset, inverse_window
 from numalogic.tools.types import artifact_t
-from numalogic.udfs import UDFFactory, StreamConf
+from numalogic.udfs import UDFFactory, StreamConf, MLPipelineConf
 
 DEFAULT_OUTPUT_DIR = os.path.join(BASE_DIR, ".btoutput")
 LOGGER = logging.getLogger(__name__)
@@ -54,8 +54,12 @@ def _init_default_streamconf(metrics: list[str]) -> StreamConf:
  return StreamConf(
  source=ConnectorType.prometheus,
  window_size=DEFAULT_SEQUENCE_LEN,
- metrics=metrics,
- numalogic_conf=numalogic_cfg,
+ ml_pipelines={
+ "default": MLPipelineConf(
+ metrics=metrics,
+ numalogic_conf=numalogic_cfg,
+ )
+ },
  )
 
 
@@ -73,6 +77,7 @@ class PromUnivarBacktester:
  output_dir: Output directory
  test_ratio: Ratio of test data to total data
  stream_conf: Stream configuration
+ pipeline_id: ml pipeline id from stream_conf
  """
 
  def __init__(
@@ -86,18 +91,20 @@ def __init__(
  output_dir: Union[str, Path] = DEFAULT_OUTPUT_DIR,
  test_ratio: float = 0.25,
  stream_conf: Optional[StreamConf] = None,
+ pipeline_id: str = "default",
  ):
  self._url = url
  self.namespace = namespace
  self.appname = appname
  self.metric = metric
  self.conf = stream_conf or _init_default_streamconf([metric])
+ self.ml_pl_conf = self.conf.ml_pipelines[pipeline_id]
  self.test_ratio = test_ratio
  self.lookback_days = lookback_days
  self.return_labels = return_labels
 
  self._seq_len = self.conf.window_size
- self._n_features = len(self.conf.metrics)
+ self._n_features = len(self.ml_pl_conf.metrics)
 
  self.out_dir = self.get_outdir(appname, metric, outdir=output_dir)
  self._datapath = os.path.join(self.out_dir, "data.csv")
@@ -170,13 +177,13 @@ def train_models(
  LOGGER.info("Training data shape: %s", x_train.shape)
 
  artifacts = UDFFactory.get_udf_cls("trainer").compute(
- model=ModelFactory().get_instance(self.conf.numalogic_conf.model),
+ model=ModelFactory().get_instance(self.ml_pl_conf.numalogic_conf.model),
  input_=x_train,
  preproc_clf=PreprocessFactory().get_pipeline_instance(
- self.conf.numalogic_conf.preprocess
+ self.ml_pl_conf.numalogic_conf.preprocess
  ),
- threshold_clf=ThresholdFactory().get_instance(self.conf.numalogic_conf.threshold),
- numalogic_cfg=self.conf.numalogic_conf,
+ threshold_clf=ThresholdFactory().get_instance(self.ml_pl_conf.numalogic_conf.threshold),
+ numalogic_cfg=self.ml_pl_conf.numalogic_conf,
  )
  artifacts_dict = {
  "model": artifacts["inference"].artifact,
@@ -233,10 +240,12 @@ def generate_scores(
 
  ds = StreamingDataset(x_scaled, seq_len=self.conf.window_size)
  anomaly_scores = np.zeros(
- (len(ds), self.conf.window_size, len(self.conf.metrics)), dtype=np.float32
+ (len(ds), self.conf.window_size, len(self.ml_pl_conf.metrics)), dtype=np.float32
  )
  x_recon = np.zeros_like(anomaly_scores, dtype=np.float32)
- postproc_func = PostprocessFactory().get_instance(self.conf.numalogic_conf.postprocess)
+ postproc_func = PostprocessFactory().get_instance(
+ self.ml_pl_conf.numalogic_conf.postprocess
+ )
 
  for idx, arr in enumerate(ds):
  x_recon[idx] = nn_udf.compute(model=artifacts["model"], input_=arr)

diff --git a/numalogic/udfs/README.md b/numalogic/udfs/README.md
@@ -1,19 +1,19 @@
-| Metric Name | Type | Labels | Description |
-|:-------------------------:|:---------:|:----------------------------------------:|:-----------------------------------------------:|
-| MSG_IN_COUNTER | Counter | vertex, composite_key, config_id | Count msgs flowing in |
-| MSG_PROCESSED_COUNTER | Counter | vertex, composite_key, config_id | Count msgs processed |
-| SOURCE_COUNTER | Counter | source, composite_key, config_id | Count artifact source (registry or cache) calls |
-| INSUFFICIENT_DATA_COUNTER | Counter | composite_key, config_id | Count insufficient data while Training |
-| MODEL_STATUS_COUNTER | Counter | status, vertex, composite_key, config_id | Count status of the model |
-| DATASHAPE_ERROR_COUNTER | Counter | composite_key, config_id | Count datashape errors in preprocess |
-| MSG_DROPPED_COUNTER | Counter | vertex, composite_key, config_id | Count dropped msgs |
-| REDIS_ERROR_COUNTER | Counter | vertex, composite_key, config_id | Count redis errors |
-| EXCEPTION_COUNTER | Counter | vertex, composite_key, config_id | Count exceptions |
-| RUNTIME_ERROR_COUNTER | Counter | vertex, composite_key, config_id | Count runtime errors |
-| FETCH_EXCEPTION_COUNTER | Counter | composite_key, config_id | Count exceptions during train data fetch calls |
-| DATAFRAME_SHAPE_SUMMARY | Summary | composite_key, config_id | len of dataframe for training |
-| NAN_SUMMARY | Summary | composite_key, config_id | Count nan's in train data |
-| INF_SUMMARY | Summary | composite_key, config_id | Count inf's in train data |
-| FETCH_TIME_SUMMARY | Summary | composite_key, config_id | Train Data Fetch time |
-| MODEL_INFO | Info | composite_key, config_id | Model info |
-| UDF_TIME | Histogram | composite_key, config_id | Histogram for udf processing time |
+| Metric Name | Type |  Labels  | Description |
+|:-------------------------:|:---------:|:-----------------------------------------------------:|:-----------------------------------------------:|
+| MSG_IN_COUNTER | Counter | vertex, composite_key, config_id, pipeline_id | Count msgs flowing in |
+| MSG_PROCESSED_COUNTER | Counter | vertex, composite_key, config_id, pipeline_id | Count msgs processed |
+| SOURCE_COUNTER | Counter | source, composite_key, config_id, pipeline_id | Count artifact source (registry or cache) calls |
+| INSUFFICIENT_DATA_COUNTER | Counter | composite_key, config_id, pipeline_id | Count insufficient data while Training |
+| MODEL_STATUS_COUNTER | Counter | status, vertex, composite_key, config_id, pipeline_id | Count status of the model |
+| DATASHAPE_ERROR_COUNTER | Counter | composite_key, config_id, pipeline_id | Count datashape errors in preprocess |
+| MSG_DROPPED_COUNTER | Counter | vertex, composite_key, config_id, pipeline_id | Count dropped msgs |
+| REDIS_ERROR_COUNTER | Counter | vertex, composite_key, config_id, pipeline_id | Count redis errors |
+| EXCEPTION_COUNTER | Counter | vertex, composite_key, config_id, pipeline_id | Count exceptions |
+| RUNTIME_ERROR_COUNTER | Counter | vertex, composite_key, config_id, pipeline_id | Count runtime errors |
+| FETCH_EXCEPTION_COUNTER | Counter | composite_key, config_id, pipeline_id | Count exceptions during train data fetch calls |
+| DATAFRAME_SHAPE_SUMMARY | Summary | composite_key, config_id, pipeline_id | len of dataframe for training |
+| NAN_SUMMARY | Summary | composite_key, config_id, pipeline_id | Count nan's in train data |
+| INF_SUMMARY | Summary | composite_key, config_id, pipeline_id | Count inf's in train data |
+| FETCH_TIME_SUMMARY | Summary | composite_key, config_id, pipeline_id | Train Data Fetch time |
+| MODEL_INFO | Info | composite_key, config_id, pipeline_id | Model info |
+| UDF_TIME | Histogram | composite_key, config_id, pipeline_id | Histogram for udf processing time |
diff --git a/numalogic/udfs/__init__.py b/numalogic/udfs/__init__.py
@@ -5,8 +5,9 @@
 
 from numalogic._constants import BASE_DIR
 from numalogic.udfs._base import NumalogicUDF
-from numalogic.udfs._config import StreamConf, PipelineConf, load_pipeline_conf
+from numalogic.udfs._config import StreamConf, PipelineConf, MLPipelineConf, load_pipeline_conf
 from numalogic.udfs.factory import UDFFactory, ServerFactory
+from numalogic.udfs.payloadtx import PayloadTransformer
 from numalogic.udfs.inference import InferenceUDF
 from numalogic.udfs.postprocess import PostprocessUDF
 from numalogic.udfs.preprocess import PreprocessUDF
@@ -25,6 +26,7 @@ def set_logger() -> None:
 
 __all__ = [
  "NumalogicUDF",
+ "PayloadTransformer",
  "PreprocessUDF",
  "InferenceUDF",
  "TrainerUDF",
@@ -34,6 +36,7 @@ def set_logger() -> None:
  "UDFFactory",
  "StreamConf",
  "PipelineConf",
+ "MLPipelineConf",
  "load_pipeline_conf",
  "ServerFactory",
  "set_logger",

diff --git a/numalogic/udfs/__main__.py b/numalogic/udfs/__main__.py
@@ -21,8 +21,12 @@ def init_server(step: str, server_type: str):
  pipeline_conf = load_pipeline_conf(BASE_CONF_FILE_PATH, APP_CONF_FILE_PATH)
  LOGGER.info("Pipeline config: %s", pipeline_conf)
 
- redis_client = get_redis_client_from_conf(pipeline_conf.redis_conf)
- udf = UDFFactory.get_udf_instance(step, r_client=redis_client, pl_conf=pipeline_conf)
+ LOGGER.info("Starting vertex with step: %s, server_type %s", step, server_type)
+ if step == "mlpipeline":
+ udf = UDFFactory.get_udf_instance(step, pl_conf=pipeline_conf)
+ else:
+ redis_client = get_redis_client_from_conf(pipeline_conf.redis_conf)
+ udf = UDFFactory.get_udf_instance(step, r_client=redis_client, pl_conf=pipeline_conf)
 
  return ServerFactory.get_server_instance(server_type, handler=udf)
 

diff --git a/numalogic/udfs/_base.py b/numalogic/udfs/_base.py
@@ -17,8 +17,7 @@
 
 from numalogic.tools.exceptions import ConfigNotFoundError
 from numalogic.tools.types import artifact_t
-from numalogic.udfs._config import PipelineConf, StreamConf
-
+from numalogic.udfs._config import StreamConf, PipelineConf, MLPipelineConf
 
 _DEFAULT_CONF_ID = "default"
 
@@ -61,17 +60,28 @@ def register_conf(self, config_id: str, conf: StreamConf) -> None:
  """
  self.pl_conf.stream_confs[config_id] = conf
 
- def _get_default_conf(self, config_id) -> StreamConf:
+ def _get_default_stream_conf(self, config_id) -> StreamConf:
  """Get the default config."""
  try:
  return self.pl_conf.stream_confs[_DEFAULT_CONF_ID]
  except KeyError:
  err_msg = f"Config with ID {config_id} or {_DEFAULT_CONF_ID} not found!"
  raise ConfigNotFoundError(err_msg) from None
 
- def get_conf(self, config_id: str) -> StreamConf:
+ def _get_default_ml_pipeline_conf(self, config_id, pipeline_id) -> MLPipelineConf:
+ """Get the default pipeline config."""
+ try:
+ return self.pl_conf.stream_confs[_DEFAULT_CONF_ID].ml_pipelines[_DEFAULT_CONF_ID]
+ except KeyError:
+ err_msg = (
+ f"Pipeline with ID {pipeline_id} or {_DEFAULT_CONF_ID} "
+ f"not found for config ID {config_id}!"
+ )
+ raise ConfigNotFoundError(err_msg) from None
+
+ def get_stream_conf(self, config_id: str) -> StreamConf:
  """
- Get config with the given ID.
+ Get stream config with the given ID.
  If not found, return the default config.
 
  Args:
@@ -88,7 +98,28 @@ def get_conf(self, config_id: str) -> StreamConf:
  try:
  return self.pl_conf.stream_confs[config_id]
  except KeyError:
- return self._get_default_conf(config_id)
+ return self._get_default_stream_conf(config_id)
+
+ def get_ml_pipeline_conf(self, config_id: str, pipeline_id: str) -> MLPipelineConf:
+ """
+ Get stream's pipeline config with the given ID.
+ If not found, return the default config.
+
+ Args:
+ config_id: Config ID
+
+ Returns
+ -------
+ StreamConf object
+
+ Raises
+ ------
+ ConfigNotFoundError: If config with the given ID is not found
+ """
+ try:
+ return self.pl_conf.stream_confs[config_id].ml_pipelines[pipeline_id]
+ except KeyError:
+ return self._get_default_ml_pipeline_conf(config_id, pipeline_id)
 
  def exec(self, keys: list[str], datum: Datum) -> Messages:
  """

diff --git a/numalogic/udfs/_config.py b/numalogic/udfs/_config.py
@@ -17,14 +17,20 @@
 _logger = logging.getLogger(__name__)
 
 
+@dataclass
+class MLPipelineConf:
+ pipeline_id: str = "default"
+ metrics: list[str] = field(default_factory=list)
+ numalogic_conf: NumalogicConf = field(default_factory=lambda: NumalogicConf())
+
+
 @dataclass
 class StreamConf:
  config_id: str = "default"
  source: ConnectorType = ConnectorType.druid # TODO: do not allow redis connector here
  window_size: int = 12
  composite_keys: list[str] = field(default_factory=list)
- metrics: list[str] = field(default_factory=list)
- numalogic_conf: NumalogicConf = field(default_factory=lambda: NumalogicConf())
+ ml_pipelines: dict[str, MLPipelineConf] = field(default_factory=dict)
 
 
 @dataclass

diff --git a/numalogic/udfs/_metrics.py b/numalogic/udfs/_metrics.py
@@ -10,74 +10,90 @@
 SOURCE_COUNTER = PromCounterMetric(
  "numalogic_artifact_source_counter",
  "Count artifact source calls",
- ["source", "vertex", "composite_key", "config_id"],
+ ["source", "vertex", "composite_key", "config_id", "pipeline_id"],
 )
 
 INSUFFICIENT_DATA_COUNTER = PromCounterMetric(
  "numalogic_insufficient_data_counter",
  "Count insufficient data while Training",
- ["composite_key", "config_id"],
+ ["composite_key", "config_id", "pipeline_id"],
 )
 MODEL_STATUS_COUNTER = PromCounterMetric(
  "numalogic_new_model_counter",
  "Count status of the model",
- ["status", "vertex", "composite_key", "config_id"],
+ ["status", "vertex", "composite_key", "config_id", "pipeline_id"],
 )
 
 DATASHAPE_ERROR_COUNTER = PromCounterMetric(
  "numalogic_datashape_error_counter",
  "Count datashape errors in preprocess",
- ["composite_key", "config_id"],
+ ["composite_key", "config_id", "pipeline_id"],
 )
 MSG_DROPPED_COUNTER = PromCounterMetric(
- "numalogic_msg_dropped_counter", "Count dropped msgs", ["vertex", "composite_key", "config_id"]
+ "numalogic_msg_dropped_counter",
+ "Count dropped msgs",
+ ["vertex", "composite_key", "config_id", "pipeline_id"],
 )
 
 REDIS_ERROR_COUNTER = PromCounterMetric(
- "numalogic_redis_error_counter", "Count redis errors", ["vertex", "composite_key", "config_id"]
+ "numalogic_redis_error_counter",
+ "Count redis errors",
+ ["vertex", "composite_key", "config_id", "pipeline_id"],
 )
 EXCEPTION_COUNTER = PromCounterMetric(
- "numalogic_exception_counter", "Count exceptions", ["vertex", "composite_key", "config_id"]
+ "numalogic_exception_counter",
+ "Count exceptions",
+ ["vertex", "composite_key", "config_id", "pipeline_id"],
 )
 RUNTIME_ERROR_COUNTER = PromCounterMetric(
  "numalogic_runtime_error_counter",
  "Count runtime errors",
- ["vertex", "composite_key", "config_id"],
+ ["vertex", "composite_key", "config_id", "pipeline_id"],
 )
 
 FETCH_EXCEPTION_COUNTER = PromCounterMetric(
  "numalogic_fetch_exception_counter",
  "count exceptions during fetch call",
- ["composite_key", "config_id"],
+ ["composite_key", "config_id", "pipeline_id"],
 )
 
 MSG_IN_COUNTER = PromCounterMetric(
- "numalogic_msg_in_counter", "Count msgs flowing in", ["vertex", "composite_key", "config_id"]
+ "numalogic_msg_in_counter",
+ "Count msgs flowing in",
+ ["vertex", "composite_key", "config_id", "pipeline_id"],
 )
 MSG_PROCESSED_COUNTER = PromCounterMetric(
  "numalogic_msg_processed_counter",
  "Count msgs processed",
- ["vertex", "composite_key", "config_id"],
+ ["vertex", "composite_key", "config_id", "pipeline_id"],
 )
 
 # SUMMARY
 DATAFRAME_SHAPE_SUMMARY = PromSummaryMetric(
  "numalogic_dataframe_shape_summary",
  "len of dataframe for training",
- ["composite_key", "config_id"],
+ ["composite_key", "config_id", "pipeline_id"],
 )
 NAN_SUMMARY = PromSummaryMetric(
- "numalogic_nan_counter", "Count nan's in train data", ["composite_key", "config_id"]
+ "numalogic_nan_summary",
+ "Count nan's in train data",
+ ["composite_key", "config_id", "pipeline_id"],
 )
 INF_SUMMARY = PromSummaryMetric(
- "numalogic_inf_counter", "Count inf's in train data", ["composite_key", "config_id"]
+ "numalogic_inf_summary",
+ "Count inf's in train data",
+ ["composite_key", "config_id", "pipeline_id"],
 )
 FETCH_TIME_SUMMARY = PromSummaryMetric(
- "numalogic_fetch_time_summary", "Train data fetch time", ["composite_key", "config_id"]
+ "numalogic_fetch_time_summary",
+ "Train data fetch time",
+ ["composite_key", "config_id", "pipeline_id"],
 )
 
 # Info
-MODEL_INFO = PromInfoMetric("numalogic_model_info", "Model info", ["composite_key", "config_id"])
+MODEL_INFO = PromInfoMetric(
+ "numalogic_model_info", "Model info", ["composite_key", "config_id", "pipeline_id"]
+)
 
 # HISTOGRAM
 buckets = (

diff --git a/numalogic/udfs/entities.py b/numalogic/udfs/entities.py
@@ -33,6 +33,7 @@ class Header(str, Enum):
 class _BasePayload:
  uuid: str
  config_id: str
+ pipeline_id: str
  composite_keys: list[str]
 
 
@@ -60,6 +61,7 @@ class StreamPayload(_BasePayload):
  timestamps: list[int]
  status: Optional[Status] = None
  header: Header = Header.MODEL_INFERENCE
+ artifact_versions: dict[str, dict] = field(default_factory=dict)
  metadata: dict[str, Any] = field(default_factory=dict)
 
  @property
@@ -85,6 +87,7 @@ def __str__(self) -> str:
  return (
  f'"StreamPayload(header={self.header}, status={self.status}, '
  f'composite_keys={self.composite_keys}, data={list(self.data)})"'
+ f"artifact_versions={self.artifact_versions}"
  )
 
  def __repr__(self) -> str: