Optimal Stopping Point for CI Tests

## Import libraries
import os
import gzip
import json
import datetime
import itertools
import scipy  # noqa F401
from scipy.stats import (  # noqa F401
    invgauss,
    lognorm,
    pearson3,
    weibull_min,
    triang,
    beta,
    norm,
    weibull_max,
    uniform,
    gamma,
    expon,
)

from ipynb.fs.defs.osp_helper_functions import (
    CephCommunication,
    fit_distribution,
    standardize,
    filter_test_type,
    fetch_all_tests,
    best_distribution,
    optimal_stopping_point,
)
import warnings

warnings.filterwarnings("ignore")

## Specify variables
METRIC_NAME = "time_to_fail"

# Specify the path for input grid data
INPUT_DATA_PATH = "../../data/raw/testgrid_258.json.gz"

# Specify the path for output metric data
OUTPUT_DATA_PATH = f"../../../../data/processed/metrics/{METRIC_NAME}"

## CEPH Bucket variables
## Create a .env file on your local with the correct configs
s3_endpoint_url = os.getenv("S3_ENDPOINT")
s3_access_key = os.getenv("S3_ACCESS_KEY")
s3_secret_key = os.getenv("S3_SECRET_KEY")
s3_bucket = os.getenv("S3_BUCKET")
s3_path = os.getenv("S3_PROJECT_KEY", "metrics")
s3_input_data_path = "raw_data"

# Specify whether or not we are running this as a notebook or part of an automation pipeline.
AUTOMATION = os.getenv("IN_AUTOMATION")

## Import data
timestamp = datetime.datetime.today()

if AUTOMATION:
    filename = f"testgrid_{timestamp.day}{timestamp.month}.json"
    cc = CephCommunication(s3_endpoint_url, s3_access_key, s3_secret_key, s3_bucket)
    s3_object = cc.s3_resource.Object(s3_bucket, f"{s3_input_data_path}/{filename}")
    file_content = s3_object.get()["Body"].read().decode("utf-8")
    testgrid_data = json.loads(file_content)

else:
    with gzip.open(INPUT_DATA_PATH, "rb") as read_file:
        testgrid_data = json.load(read_file)

# Fetch all failing tests i.e which have a status code of 12
failures_df = fetch_all_tests(testgrid_data, 12)

failures_df.head()

# Fetch all passing tests i.e which have a status code of 1
passing_df = fetch_all_tests(testgrid_data, 1)

passing_df.head()

failures_test = filter_test_type(
    failures_df,
    "operator.Run multi-stage test e2e-aws-upgrade - e2e-aws-upgrade-openshift-e2e-test container test",
)
failures_test.head()

passing_test = filter_test_type(
    passing_df,
    "operator.Run multi-stage test e2e-aws-upgrade - e2e-aws-upgrade-openshift-e2e-test container test",
)
passing_test.head()

failure_dist, failures_r = fit_distribution(failures_test, "test_duration", 0.99, 0.01)

# Identify the best fit distribution from the failing test along with its corresponding distribution parameters
best_dist, parameters_failing = best_distribution(failure_dist, failures_r)

# Identify the distributions for the passing test along with its corresponding distribution parameters
passing_dist, passing_r = fit_distribution(passing_test, "test_duration", 0.99, 0.01)

passing_r.head()

# Identify the best fit distribution from the passing test
best_distribution(passing_dist, passing_r)

# Find the corresponding passing test distribution parameters for the
# best fit distribution identified from the failing test above
parameters_passing = passing_dist[passing_dist["Distribution Names"] == best_dist][
    "Parameters"
].values
parameters_passing = list(itertools.chain(*parameters_passing))

# Standardize the features by removing the mean and scaling to unit variance
y_std_failing, len_y_failing, y_failing = standardize(
    failures_test, "test_duration", 0.99, 0.01
)

# Standardize the features by removing the mean and scaling to unit variance
y_std_passing, len_y_passing, y_passing = standardize(
    passing_test, "test_duration", 0.99, 0.01
)

osp = optimal_stopping_point(
    best_dist,
    y_std_failing,
    y_failing,
    parameters_failing,
    y_std_passing,
    y_passing,
    parameters_passing,
)

# Optimat Stopping Point for `operator.Run multi-stage test e2e-aws-upgrade
# - e2e-aws-upgrade-openshift-e2e-test container test`
osp