Observe Azure ML Pipelines in New Relic

Introduction

30 Second Summary

An overnight data job can fail after processing thousands of records. A bare error leaves you searching through the run for the broken row.

In this project, you will instrument a Python order-processing component with OpenTelemetry. Its Azure Machine Learning run exports searchable logs and metrics to New Relic.

What You'll Build

You will watch a malformed order run finish with a usable summary while one correlation ID links its failed-row log to its health metrics in New Relic.

By the end of this project, you'll have:

• A working local component where a valid CSV produces a summary showing the processed record count and total order amount.
• Structured logs and custom metrics that keep three valid records moving while identifying order A103 as the failed row. The same telemetry reports one failure plus the component duration.
• A secure remote pipeline run that retrieves its ingest license key from Azure Key Vault. Its logs and metrics remain searchable under one correlation ID.
• Secret Mission: A static NRQL alert condition that surfaces failed pipeline records before someone manually searches the telemetry.

Are there any prerequisites?

You need an Azure subscription with an existing Azure Machine Learning workspace. You also need a New Relic account with an ingest license key.

Your environment needs Python 3.10 or later plus an editor. Your Azure account needs permission to create compute and access the workspace's Azure Key Vault secrets.

Before We Start

This checkpoint locks in the production pipeline component that deserves code-level observability. You will target an operational failure with OpenTelemetry signals that complement Azure Monitor platform monitoring.

Set Up the Local Project

Reliable observability depends on a reproducible Python environment. Loose global installs can give each machine a different SDK surface.

This step creates the local layout used by the batch component. The dependency files pin the Azure Machine Learning and OpenTelemetry packages before any pipeline code is written.

In this step, get ready to:

• Prepare the local project structure.
• Install the pinned Python dependencies.
• Verify the Azure Machine Learning and OpenTelemetry imports.

Create the local project structure

• Launch your editor from your operating system's app search.
• Use your editor's folder controls to create a directory named azureml-otel on your Desktop.

Your editor should now show azureml-otel as the open directory.

• Create a data folder inside azureml-otel with your editor's folder control.
• Create a src folder inside azureml-otel with the same control.

You should see data and src beneath azureml-otel in the file sidebar.

• Create an outputs folder inside azureml-otel with your editor's folder control.

The file sidebar should now show the data, src, and outputs folders.

• Create requirements.txt directly inside azureml-otel with your editor's new-file control.

You should see an empty requirements.txt beside the three folders.

• Populate requirements.txt by pasting this content:

azure-ai-ml==1.35.2
azure-identity==1.27.0
azure-keyvault-secrets==4.12.0
opentelemetry-api==1.45.2
opentelemetry-sdk==1.45.2
opentelemetry-exporter-otlp-proto-http==1.45.2
opentelemetry-instrumentation-logging==0.66b2

Why pin every dependency?

• Each == pin selects one package release.
• azure-ai-ml provides the remote pipeline SDK.
• azure-identity provides the authentication support used later.
• azure-keyvault-secrets provides secure secret access.
• The OpenTelemetry packages provide the pinned telemetry stack.

• Save requirements.txt.
• Confirm that the file contains seven pinned package lines.

Does the package file look wrong?

• Check that requirements.txt sits directly inside azureml-otel.
• Remove any blank package lines inserted between the seven entries.

Help me check my requirements file.

• Create environment.yml directly inside azureml-otel with your editor's new-file control.

You should see an empty environment.yml beside requirements.txt.

• Populate environment.yml by pasting this content:

name: azureml-otel
channels:
  - conda-forge
dependencies:
  - python=3.10
  - pip
  - pip:
      - azure-ai-ml==1.35.2
      - azure-identity==1.27.0
      - azure-keyvault-secrets==4.12.0
      - opentelemetry-api==1.45.2
      - opentelemetry-sdk==1.45.2
      - opentelemetry-exporter-otlp-proto-http==1.45.2
      - opentelemetry-instrumentation-logging==0.66b2

What does this file do?

• name: azureml-otel labels the remote environment.
• python=3.10 fixes the runtime used by the pipeline job.
• conda-forge supplies the Conda packages.
• The nested pip section reproduces the local package set on Azure compute.

• Save environment.yml.
• Confirm that the indentation matches the displayed file.

Does the environment file show a warning?

• Replace tabs with spaces throughout environment.yml.
• Check that every package entry under the nested pip key has six leading spaces.

Help me fix my environment file.

Install the pinned dependencies

The local installation must match the package versions recorded in requirements.txt. The single install command keeps that package set consistent.

• Open your editor's terminal with azureml-otel as the current directory.
• Install the pinned packages by running this command:

pip install azure-ai-ml==1.35.2 azure-identity==1.27.0 azure-keyvault-secrets==4.12.0 opentelemetry-api==1.45.2 opentelemetry-sdk==1.45.2 opentelemetry-exporter-otlp-proto-http==1.45.2 opentelemetry-instrumentation-logging==0.66b2

What does this command do?

pip installs each package at its recorded version. This gives the local environment the Azure SDK and OpenTelemetry modules required by the project.

• Wait for the command to return to the terminal prompt.
• Confirm that the output finishes without an installation failure.

Did the package installation fail?

• Confirm that the active Python environment uses version 3.10 or higher.
• Check that your network allows package downloads from PyPI.
• Rerun the command after correcting the reported package or network issue.

Help me diagnose the dependency installation.

Add the sample data and verify imports

The project needs one clean dataset and one malformed dataset. These files create a controlled comparison for the baseline processor in the next step.

• Create orders.csv inside the data folder with your editor's new-file control.

You should see an empty data/orders.csv in the file sidebar.

• Populate data/orders.csv by pasting this content:

order_id,amount
A100,19.95
A101,42.50
A102,8.25

What does this data contain?

• order_id identifies each sample order.
• amount holds the numeric value processed by the component.
• The three rows provide valid input for the baseline run.

• Save data/orders.csv.
• Confirm that the file shows one header row plus three order rows.

Does the valid CSV look misaligned?

• Check that each row contains one comma.
• Remove any spaces surrounding the commas.

Help me check the valid CSV.

• Create orders_bad.csv inside the data folder with your editor's new-file control.

You should see an empty data/orders_bad.csv beside the valid dataset.

• Populate data/orders_bad.csv by pasting this content:

order_id,amount
A100,19.95
A101,42.50
A102,8.25
A103,not-a-number

What makes this data malformed?

• The first three records preserve the valid baseline.
• A103 uses not-a-number where the processor expects a numeric amount.
• The fourth record provides the controlled failure input for the next step.

• Save data/orders_bad.csv.
• Confirm that the file shows one header row plus four order rows.

Is the malformed row missing?

• Check that A103,not-a-number is the final row.
• Confirm that the filename is orders_bad.csv.

Help me check the malformed CSV.

Use the tabs below to compare all four project files before checking the installed imports.

✔️ Awesome, I've got everything!

Great. Your four project files match the required starting state.

ⓧ I'd like to double check the full code

• Compare each file below with the matching file in your editor.

requirements.txt

azure-ai-ml==1.35.2
azure-identity==1.27.0
azure-keyvault-secrets==4.12.0
opentelemetry-api==1.45.2
opentelemetry-sdk==1.45.2
opentelemetry-exporter-otlp-proto-http==1.45.2
opentelemetry-instrumentation-logging==0.66b2

environment.yml

name: azureml-otel
channels:
  - conda-forge
dependencies:
  - python=3.10
  - pip
  - pip:
      - azure-ai-ml==1.35.2
      - azure-identity==1.27.0
      - azure-keyvault-secrets==4.12.0
      - opentelemetry-api==1.45.2
      - opentelemetry-sdk==1.45.2
      - opentelemetry-exporter-otlp-proto-http==1.45.2
      - opentelemetry-instrumentation-logging==0.66b2

data/orders.csv

order_id,amount
A100,19.95
A101,42.50
A102,8.25

data/orders_bad.csv

order_id,amount
A100,19.95
A101,42.50
A102,8.25
A103,not-a-number

Before you run the final check, do you expect both imports to load from the environment you just prepared?

• Verify both installed imports by running this command:

python -c "import azure.ai.ml, opentelemetry.sdk; print('dependencies ready')"

What does this check prove?

• import azure.ai.ml loads the Azure Machine Learning SDK.
• import opentelemetry.sdk loads the OpenTelemetry SDK.
• print('dependencies ready') runs only after both imports succeed.

You should see dependencies ready in the terminal. That output confirms the installed environment can load both SDKs.

Are the imports failing?

• Confirm that the terminal uses the same Python environment where pip installed the packages.
• Rerun the installation command if the output mentions a missing module.

Help me fix the import check.

Your reproducible local project is ready. Next, you will build the plain batch processor. You will then watch malformed data break it.

Build the Fail-Fast Baseline

Your pinned environment now gives every run the same dependency foundation. The sample order files give you a controlled way to compare successful input with malformed input.

A plain batch script can look adequate until one malformed record stops the component. This step exposes how little searchable run context that failure provides.

In this step, get ready to:

• Define the command-line contract for the order processor.
• Process valid order rows into a JSON summary.
• Run malformed input to expose the missing failure context.

Define the baseline script

The script needs a consistent interface for its input file. It also needs values that identify the output location plus the current run.

• Select the existing src folder in your editor's file sidebar.
• Create process_data.py inside the src folder.
• Add the imports plus the argument parser by pasting this code into src/process_data.py:

from __future__ import annotations

import argparse
import csv
import json
from pathlib import Path
from time import perf_counter
from typing import Any


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output-dir", required=True)
    parser.add_argument("--run-id", required=True)
    parser.add_argument(
        "--exporter-mode",
        choices=("off", "console", "otlp"),
        required=True,
    )
    parser.add_argument("--key-vault-url")
    parser.add_argument("--secret-name")
    parser.add_argument("--otlp-endpoint")
    return parser.parse_args()

What does this code do?

• The imports provide command-line parsing plus CSV reading.
• The parse_args() function requires an input path plus an output directory.
• The run identifier gives each execution a distinct label.
• The exporter mode determines which processing path runs.

• Save src/process_data.py.

You should see parse_args in your editor's symbol outline or function list.

Can't see the argument parser?

Confirm that process_data.py sits inside the existing src folder. Check that the function begins at the left edge of the file.

Ask for help with the parser structure:

A batch component also needs a durable result after successful processing. The summary writer creates the output directory before saving a formatted JSON artifact.

• Place the next function directly below parse_args() by adding this code:

def write_summary(output_dir: str, summary: dict[str, Any]) -> Path:
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    summary_path = output_path / "summary.json"
    summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8")
    print(f"Summary written to {summary_path}")
    return summary_path

How does the summary writer help?

• The output directory is created when it does not already exist.
• The result is written to a predictable summary.json path.
• The printed path gives the terminal a visible success signal.

• Save src/process_data.py.

You should now see write_summary listed below parse_args in your editor's symbol outline.

Is the summary function nested?

Check that def write_summary begins at the left edge of the file. An extra indent can accidentally place it inside parse_args().

Ask for help checking the function placement:

The baseline processor treats every amount as valid numeric input. A conversion failure immediately escapes the function because there is no per-record error boundary.

• Place the fail-fast processor directly below write_summary() by adding this code:

def process_fail_fast(input_path: str, output_dir: str, run_id: str) -> None:
    processed = 0
    total_amount = 0.0

    with Path(input_path).open(newline="", encoding="utf-8") as source:
        for row in csv.DictReader(source):
            amount = float(row["amount"])
            if amount < 0:
                raise ValueError("amount must be non-negative")
            processed += 1
            total_amount += amount

    write_summary(
        output_dir,
        {
            "run_id": run_id,
            "processed": processed,
            "failed": 0,
            "total_amount": round(total_amount, 2),
        },
    )

What does the fail-fast path do?

• The csv.DictReader maps each row to the order_id plus amount columns.
• The float() conversion rejects an amount that is not numeric.
• Valid amounts increase the processed count plus the running total.
• The completed values pass to write_summary() only after every row succeeds.

• Save src/process_data.py.

You should now see process_fail_fast as the third function in your editor's symbol outline.

Does the processor look incomplete?

Confirm that write_summary() sits after the file-reading block. Check that it remains inside process_fail_fast().

Ask for help checking the baseline flow:

The final entry point connects parsed arguments to the selected processing path. The off mode chooses the fail-fast baseline used in this step.

• Place the script entry point directly below process_fail_fast() by adding this code:

def main() -> None:
    args = parse_args()
    if args.exporter_mode == "off":
        process_fail_fast(args.input, args.output_dir, args.run_id)
    else:
        process_observed(args)


if __name__ == "__main__":
    main()

How does the entry point work?

• The main() function collects the command-line values.
• The off value sends the input through process_fail_fast().
• The module guard starts main() when Python runs this file directly.

• Save src/process_data.py.

Your file now has an executable baseline path for off mode.

Seeing an unresolved function warning?

Your editor may flag process_observed() until you reconcile the cumulative file after the designed failure. The off branch still selects the complete baseline path.

Ask for help checking the execution branch:

Prove the valid path

The valid sample contains three numeric amounts. A successful run should reach the summary writer after processing every row.

Before you run this, consider whether all three amounts can reach the summary writer.

• Run the valid-data baseline from your existing local project directory with this command:

python src/process_data.py --input data/orders.csv --output-dir outputs/baseline --run-id local-baseline --exporter-mode off

What should you see?

The terminal prints a message naming outputs/baseline/summary.json. This confirms that every valid row reached the summary writer.

• Expand the existing outputs folder in your editor's file sidebar.
• Expand the new baseline folder.
• Select summary.json.

You should see the run identifier local-baseline. You should also see 3 processed records plus a total amount of 70.7.

No summary file?

Confirm that your terminal is positioned inside the local project directory containing data plus src. Check that the valid input path uses data/orders.csv.

Ask for help with the valid baseline run:

Good progress. Your baseline now turns three valid orders into a durable summary artifact.

Expose the malformed-row failure

The second sample adds order A103 with the amount not-a-number. The baseline has no row-level exception handling or structured telemetry.

Before you run this, consider whether one malformed amount stops only its row or the entire component.

• Run the malformed-data baseline from your existing local project directory with this command:

python src/process_data.py --input data/orders_bad.csv --output-dir outputs/baseline-bad --run-id local-baseline-bad --exporter-mode off

This failure is intentional

The terminal exits with a ValueError mentioning not-a-number. The exception escapes as soon as Python tries to convert the malformed amount.

The traceback lacks a searchable run identifier plus business context for order A103. That gap is the designed obstacle.

• Inspect the final lines of the traceback.
• Check the outputs/baseline-bad location in your editor's file sidebar.

You should see no summary.json artifact for this failed run. The malformed fourth row prevents the component from recording the three valid rows that came before it.

Did the malformed run succeed?

Confirm that the command uses data/orders_bad.csv. Check that its final row contains A103 plus not-a-number.

Ask for help reproducing the intended failure:

The failure has now made the observability gap visible. Reconcile the cumulative file so its function names plus execution branches match the project source of truth.

• Compare src/process_data.py with the complete reference below.

✔️ Awesome, I've got everything!

• Save src/process_data.py.

Your cumulative file now contains the fail-fast baseline plus the observed execution boundary used next.

ⓧ I'd like to double check the full code

• Use this reference to correct every mismatch in src/process_data.py.

from __future__ import annotations

import argparse
import csv
import json
from pathlib import Path
from time import perf_counter
from typing import Any


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--output-dir", required=True)
    parser.add_argument("--run-id", required=True)
    parser.add_argument(
        "--exporter-mode",
        choices=("off", "console", "otlp"),
        required=True,
    )
    parser.add_argument("--key-vault-url")
    parser.add_argument("--secret-name")
    parser.add_argument("--otlp-endpoint")
    return parser.parse_args()


def write_summary(output_dir: str, summary: dict[str, Any]) -> Path:
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    summary_path = output_path / "summary.json"
    summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8")
    print(f"Summary written to {summary_path}")
    return summary_path


def process_fail_fast(input_path: str, output_dir: str, run_id: str) -> None:
    processed = 0
    total_amount = 0.0

    with Path(input_path).open(newline="", encoding="utf-8") as source:
        for row in csv.DictReader(source):
            amount = float(row["amount"])
            if amount < 0:
                raise ValueError("amount must be non-negative")
            processed += 1
            total_amount += amount

    write_summary(
        output_dir,
        {
            "run_id": run_id,
            "processed": processed,
            "failed": 0,
            "total_amount": round(total_amount, 2),
        },
    )


def process_observed(args: argparse.Namespace) -> None:
    from telemetry import configure_telemetry

    telemetry = configure_telemetry(
        exporter_mode=args.exporter_mode,
        run_id=args.run_id,
        key_vault_url=args.key_vault_url,
        secret_name=args.secret_name,
        otlp_endpoint=args.otlp_endpoint,
    )
    started = perf_counter()
    processed = 0
    failed = 0
    total_amount = 0.0
    outcome = "completed"
    common_attributes = {
        "pipeline.run.id": args.run_id,
        "pipeline.component": "order-processor",
    }

    telemetry.logger.info("pipeline_started", extra=common_attributes)

    try:
        with Path(args.input).open(newline="", encoding="utf-8") as source:
            for row_number, row in enumerate(csv.DictReader(source), start=2):
                record_attributes = {
                    **common_attributes,
                    "order.id": row.get("order_id", "unknown"),
                    "record.row": row_number,
                }
                try:
                    amount = float(row["amount"])
                    if amount < 0:
                        raise ValueError("amount must be non-negative")
                    processed += 1
                    total_amount += amount
                    telemetry.records_processed.add(1, attributes=record_attributes)
                    telemetry.logger.info(
                        "record_processed",
                        extra={**record_attributes, "order.amount": amount},
                    )
                except (KeyError, TypeError, ValueError):
                    failed += 1
                    telemetry.records_failed.add(1, attributes=record_attributes)
                    telemetry.logger.exception(
                        "record_failed", extra=record_attributes
                    )

        write_summary(
            args.output_dir,
            {
                "run_id": args.run_id,
                "processed": processed,
                "failed": failed,
                "total_amount": round(total_amount, 2),
            },
        )
    except Exception:
        outcome = "failed"
        telemetry.logger.exception("pipeline_failed", extra=common_attributes)
        raise
    finally:
        duration_seconds = perf_counter() - started
        telemetry.component_duration.record(
            duration_seconds,
            attributes={**common_attributes, "pipeline.status": outcome},
        )
        telemetry.logger.info(
            "pipeline_finished",
            extra={
                **common_attributes,
                "pipeline.status": outcome,
                "pipeline.records.processed": processed,
                "pipeline.records.failed": failed,
                "pipeline.duration.seconds": duration_seconds,
            },
        )
        telemetry.close()


def main() -> None:
    args = parse_args()
    if args.exporter_mode == "off":
        process_fail_fast(args.input, args.output_dir, args.run_id)
    else:
        process_observed(args)


if __name__ == "__main__":
    main()

How is the cumulative file organized?

• The off branch preserves the fail-fast behavior you just observed.
• The process_observed() boundary keeps the next instrumentation changes isolated from the baseline.
• The shared write_summary() function keeps both paths aligned on one output format.

Before the final check, consider whether the cumulative file changes the behavior of off mode.

• Rerun the malformed-data baseline with this command:

python src/process_data.py --input data/orders_bad.csv --output-dir outputs/baseline-bad --run-id local-baseline-bad --exporter-mode off

What does the final check prove?

The terminal still exits with a ValueError mentioning not-a-number. The off mode remains a clean before-state for the observability work ahead.

You now have a successful baseline plus a reproducible failure with weak context. Next, you will add structured logs plus metrics that preserve valid work and identify the rejected row.

Add Structured Telemetry

Your fail-fast baseline exposed the malformed amount in A103. The crash gave you no searchable run context or row context.

This step adds a reusable OpenTelemetry boundary that records failures without stopping valid rows. It also flushes short-lived batch telemetry before the process exits.

In this step, get ready to:

• Build a reusable telemetry adapter with isolated providers.
• Configure console or OTLP export for operational signals.
• Run the malformed data through the observed processing path.

Build the telemetry adapter

The adapter keeps telemetry setup separate from order-processing logic. This separation lets process_observed() use the same logger and instruments regardless of the selected exporter.

• In the Explorer sidebar of your editor, create telemetry.py inside the src folder.
• Add the telemetry dependencies by pasting this import group into src/telemetry.py:

from __future__ import annotations

import logging
import os
from dataclasses import dataclass
from typing import Any

from azure.ai.ml.identity import AzureMLOnBehalfOfCredential
from azure.keyvault.secrets import SecretClient
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.instrumentation.logging.handler import LoggingHandler
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk._logs.export import (
    BatchLogRecordProcessor,
    ConsoleLogRecordExporter,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    ConsoleMetricExporter,
    PeriodicExportingMetricReader,
)
from opentelemetry.sdk.resources import Resource

What do these imports provide?

• The Azure imports provide job identity plus secure access to Azure Key Vault.
• The exporter imports support console output plus OTLP delivery.
• The provider imports create isolated logging and metric pipelines for this component.

• Save src/telemetry.py.
• Open the editor's Problems panel.

You should see no syntax errors in the import group.

Seeing unresolved imports?

Confirm that your editor uses the Python environment where the pinned dependencies were installed. Check that each import matches the block exactly.

Help me diagnose unresolved imports in telemetry.py.

A short-lived batch process needs one object that owns every telemetry resource. The Telemetry class also gives the component one reliable cleanup method.

• Add this data class below the import group in src/telemetry.py:

@dataclass
class Telemetry:
    logger: logging.Logger
    meter_provider: MeterProvider
    logger_provider: LoggerProvider
    log_handler: LoggingHandler
    records_processed: Any
    records_failed: Any
    component_duration: Any

    def close(self) -> None:
        self.logger.removeHandler(self.log_handler)
        self.logger_provider.force_flush()
        self.meter_provider.force_flush()
        self.logger_provider.shutdown()
        self.meter_provider.shutdown()

What does the telemetry container do?

• The fields hold the isolated providers plus the logger used by the component.
• The instrument fields expose counters plus a duration histogram to the processing code.
• The close() method forces buffered signals out before shutting both providers down.

• Save src/telemetry.py.
• Open the editor's symbol outline for src/telemetry.py.

You should see Telemetry with a nested close method.

Class missing from the outline?

Check that @dataclass starts at the left edge. Keep every line inside close() indented by eight spaces.

Help me fix the structure of my Telemetry class.

Remote export needs an authentication header without exposing the ingest credential in source code. The helper retrieves the license key from the workspace-linked vault at runtime.

• Add this helper below the Telemetry class:

def _new_relic_headers(key_vault_url: str, secret_name: str) -> dict[str, str]:
    credential = AzureMLOnBehalfOfCredential()
    try:
        with SecretClient(vault_url=key_vault_url, credential=credential) as client:
            license_key = client.get_secret(secret_name).value
    finally:
        credential.close()

    if not license_key:
        raise RuntimeError(f"Key Vault secret {secret_name!r} has no value")
    return {"api-key": license_key}

How does secure retrieval work?

• Azure Machine Learning supplies the job submitter's identity through AzureMLOnBehalfOfCredential.
• SecretClient reads the named secret from the vault.
• The helper returns the api-key header required by New Relic.

• Save src/telemetry.py.
• Return to the editor's symbol outline.

You should now see _new_relic_headers beneath the Telemetry class.

Helper missing from the outline?

Make sure def _new_relic_headers starts at the left edge. Confirm that its try block has a matching finally block.

Help me correct my _new_relic_headers helper.

Configure the providers and instruments

The configuration function builds one resource identity for every signal. That shared identity makes logs plus metrics searchable under the same run.

This function is the fiddliest part because each new block must remain inside configure_telemetry(). Keep the indentation shown in each snippet.

• Add the function signature plus resource attributes below _new_relic_headers():

def configure_telemetry(
    *,
    exporter_mode: str,
    run_id: str,
    key_vault_url: str | None = None,
    secret_name: str | None = None,
    otlp_endpoint: str | None = None,
) -> Telemetry:
    resource = Resource.create(
        {
            "service.name": "azureml-order-processor",
            "service.version": "1.0.0",
            "pipeline.run.id": run_id,
            "pipeline.component": "order-processor",
        }
    )

Why use resource attributes?

The resource identifies the emitting service plus its pipeline run. Every exported log or metric inherits these attributes.

• Save src/telemetry.py.
• Open the editor's symbol outline.

You should see configure_telemetry as a third top-level symbol.

Function missing from the outline?

Check the closing parenthesis after Resource.create(). Confirm that the function definition starts at the left edge.

Help me fix the opening structure of configure_telemetry().

Exporter mode controls where the signals go. Console mode creates local proof while OTLP mode prepares the same instrumentation for remote delivery.

• Place your cursor after the Resource.create() block inside configure_telemetry().
• Add the exporter selection block with the indentation shown below:

    if exporter_mode == "console":
        log_exporter = ConsoleLogRecordExporter()
        metric_exporter = ConsoleMetricExporter()
    elif exporter_mode == "otlp":
        if not key_vault_url or not secret_name or not otlp_endpoint:
            raise ValueError(
                "OTLP mode requires key_vault_url, secret_name, and otlp_endpoint"
            )
        os.environ.setdefault(
            "OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE", "delta"
        )
        os.environ.setdefault(
            "OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION",
            "base2_exponential_bucket_histogram",
        )
        headers = _new_relic_headers(key_vault_url, secret_name)
        base_endpoint = otlp_endpoint.rstrip("/")
        log_exporter = OTLPLogExporter(
            endpoint=f"{base_endpoint}/v1/logs", headers=headers
        )
        metric_exporter = OTLPMetricExporter(
            endpoint=f"{base_endpoint}/v1/metrics", headers=headers
        )
    else:
        raise ValueError(f"Unsupported exporter mode: {exporter_mode}")

How does exporter selection work?

• Console mode prints structured signals locally for immediate inspection.
• OTLP mode retrieves the credential before creating signal-specific exporters.
• The final branch rejects unsupported modes before processing starts.

• Save src/telemetry.py.
• Open the editor's Problems panel.

You should see no indentation or syntax errors in the exporter branches.

Seeing an indentation error?

Keep the first if exporter_mode line indented inside the function. Align elif plus else with that first branch.

Help me repair the exporter branch indentation in configure_telemetry().

The providers turn exporters into active telemetry pipelines. The isolated logger avoids changing global logging behavior elsewhere in the project.

• Place your cursor after the unsupported-mode branch inside configure_telemetry().
• Add the provider plus logger configuration below:

    metric_reader = PeriodicExportingMetricReader(
        metric_exporter,
        export_interval_millis=5_000,
    )
    meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
    meter = meter_provider.get_meter("nextwork.azureml.observability")

    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(
        BatchLogRecordProcessor(log_exporter, schedule_delay_millis=1_000)
    )
    log_handler = LoggingHandler(
        level=logging.INFO,
        logger_provider=logger_provider,
        log_code_attributes=True,
    )
    logger = logging.getLogger("azureml.observable_order_processor")
    logger.handlers.clear()
    logger.addHandler(log_handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False

What do the providers control?

• The metric reader periodically sends measurements through the selected metric exporter.
• The batch processor buffers log records before passing them to the selected log exporter.
• The named logger uses only its dedicated handler because propagation is disabled.

• Save src/telemetry.py.
• Open the editor's Problems panel.

You should see no new syntax errors after the provider configuration.

Provider names unresolved?

Compare the provider names with the import group at the top of the file. Confirm that this block remains indented inside configure_telemetry().

Help me diagnose unresolved provider names in configure_telemetry().

The final return packages the providers with three instruments. The processing path can then count valid rows, count rejected rows, plus record total duration.

• Place your cursor after logger.propagate = False inside configure_telemetry().
• Complete the function with this return block:

    return Telemetry(
        logger=logger,
        meter_provider=meter_provider,
        logger_provider=logger_provider,
        log_handler=log_handler,
        records_processed=meter.create_counter(
            "pipeline.records.processed",
            unit="1",
            description="Valid records processed by the component",
        ),
        records_failed=meter.create_counter(
            "pipeline.records.failed",
            unit="1",
            description="Records rejected by the component",
        ),
        component_duration=meter.create_histogram(
            "pipeline.component.duration",
            unit="s",
            description="Component execution duration",
        ),
    )

What does the adapter expose?

• pipeline.records.processed counts valid records.
• pipeline.records.failed counts rejected records.
• pipeline.component.duration records component execution time in seconds.

• Save src/telemetry.py.
• Open the editor's Problems panel.

You should see no syntax errors across the completed telemetry adapter.

Return block showing errors?

Keep the return block indented inside configure_telemetry(). Check that every instrument call has its closing parenthesis.

Help me fix the instrument return block in telemetry.py.

✔️ Awesome, I've got everything!

Great work. Save src/telemetry.py before running the observed processing path.

ⓧ I'd like to double check the full code

from __future__ import annotations

import logging
import os
from dataclasses import dataclass
from typing import Any

from azure.ai.ml.identity import AzureMLOnBehalfOfCredential
from azure.keyvault.secrets import SecretClient
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.instrumentation.logging.handler import LoggingHandler
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk._logs.export import (
    BatchLogRecordProcessor,
    ConsoleLogRecordExporter,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    ConsoleMetricExporter,
    PeriodicExportingMetricReader,
)
from opentelemetry.sdk.resources import Resource


@dataclass
class Telemetry:
    logger: logging.Logger
    meter_provider: MeterProvider
    logger_provider: LoggerProvider
    log_handler: LoggingHandler
    records_processed: Any
    records_failed: Any
    component_duration: Any

    def close(self) -> None:
        self.logger.removeHandler(self.log_handler)
        self.logger_provider.force_flush()
        self.meter_provider.force_flush()
        self.logger_provider.shutdown()
        self.meter_provider.shutdown()


def _new_relic_headers(key_vault_url: str, secret_name: str) -> dict[str, str]:
    credential = AzureMLOnBehalfOfCredential()
    try:
        with SecretClient(vault_url=key_vault_url, credential=credential) as client:
            license_key = client.get_secret(secret_name).value
    finally:
        credential.close()

    if not license_key:
        raise RuntimeError(f"Key Vault secret {secret_name!r} has no value")
    return {"api-key": license_key}


def configure_telemetry(
    *,
    exporter_mode: str,
    run_id: str,
    key_vault_url: str | None = None,
    secret_name: str | None = None,
    otlp_endpoint: str | None = None,
) -> Telemetry:
    resource = Resource.create(
        {
            "service.name": "azureml-order-processor",
            "service.version": "1.0.0",
            "pipeline.run.id": run_id,
            "pipeline.component": "order-processor",
        }
    )

    if exporter_mode == "console":
        log_exporter = ConsoleLogRecordExporter()
        metric_exporter = ConsoleMetricExporter()
    elif exporter_mode == "otlp":
        if not key_vault_url or not secret_name or not otlp_endpoint:
            raise ValueError(
                "OTLP mode requires key_vault_url, secret_name, and otlp_endpoint"
            )
        os.environ.setdefault(
            "OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE", "delta"
        )
        os.environ.setdefault(
            "OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION",
            "base2_exponential_bucket_histogram",
        )
        headers = _new_relic_headers(key_vault_url, secret_name)
        base_endpoint = otlp_endpoint.rstrip("/")
        log_exporter = OTLPLogExporter(
            endpoint=f"{base_endpoint}/v1/logs", headers=headers
        )
        metric_exporter = OTLPMetricExporter(
            endpoint=f"{base_endpoint}/v1/metrics", headers=headers
        )
    else:
        raise ValueError(f"Unsupported exporter mode: {exporter_mode}")

    metric_reader = PeriodicExportingMetricReader(
        metric_exporter,
        export_interval_millis=5_000,
    )
    meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
    meter = meter_provider.get_meter("nextwork.azureml.observability")

    logger_provider = LoggerProvider(resource=resource)
    logger_provider.add_log_record_processor(
        BatchLogRecordProcessor(log_exporter, schedule_delay_millis=1_000)
    )
    log_handler = LoggingHandler(
        level=logging.INFO,
        logger_provider=logger_provider,
        log_code_attributes=True,
    )
    logger = logging.getLogger("azureml.observable_order_processor")
    logger.handlers.clear()
    logger.addHandler(log_handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False

    return Telemetry(
        logger=logger,
        meter_provider=meter_provider,
        logger_provider=logger_provider,
        log_handler=log_handler,
        records_processed=meter.create_counter(
            "pipeline.records.processed",
            unit="1",
            description="Valid records processed by the component",
        ),
        records_failed=meter.create_counter(
            "pipeline.records.failed",
            unit="1",
            description="Records rejected by the component",
        ),
        component_duration=meter.create_histogram(
            "pipeline.component.duration",
            unit="s",
            description="Component execution duration",
        ),
    )

How should the full file be structured?

The file contains an import group, the telemetry container, the secure header helper, plus the configuration function. Compare the order plus indentation with your saved file.

Run the malformed data with console telemetry

The existing process_observed() path already catches row-level conversion failures. It attaches the run ID, component name, order ID, plus CSV row number before continuing.

Before you run the observed path, do you expect the malformed row to stop the whole component?

• Run the malformed input through the console exporter by running this command:

python src/process_data.py --input data/orders_bad.csv --output-dir outputs/observed --run-id local-observed --exporter-mode console

What should you see?

The component continues after logging record_failed for order A103. The console output also contains pipeline.records.failed plus the duration metric.

The final summary reports three processed records plus one failed record. The telemetry flushes during Telemetry.close() before the process exits.

• Switch back to your editor.
• Select outputs/observed/summary.json in the Explorer sidebar.

You should see local-observed as the run ID. You should also see three processed records plus one failed record.

Missing structured output?

Confirm that src/telemetry.py is saved beside src/process_data.py. Check that you used console as the exporter mode.

Help me troubleshoot missing record_failed telemetry.

That closes the observability gap from the baseline. Your component now preserves valid work while reporting the exact malformed record.

Next, you will run this same instrumented component on Azure Machine Learning compute with secure remote export.

Submit the Azure ML Pipeline

Your local processor now turns malformed rows into structured OpenTelemetry signals. The next challenge is carrying those signals onto remote compute without exposing the New Relic ingest key.

This step submits the component to Azure Machine Learning. Azure Key Vault protects the key while the remote job exports through OTLP/HTTP.

A unique correlation ID follows the run from submission to telemetry. That identifier gives you one handle for finding the remote job later.

In this step, get ready to:

• Build repeatable submission and cleanup scripts for the remote pipeline.
• Authenticate through your browser and confirm access to the workspace Key Vault.
• Run the pipeline on autoscaling compute and capture its correlation ID.

Build the remote submission scripts

The submission script packages the security boundary around your component. It stores the ingest key before remote compute receives any job configuration.

• Create submit_pipeline.py beside requirements.txt by using your editor's file sidebar.
• Paste this foundation into submit_pipeline.py:

from __future__ import annotations

import argparse
import getpass
import uuid
from pathlib import Path

from azure.ai.ml import Input, MLClient, Output, command, dsl
from azure.ai.ml.entities import AmlCompute, Environment, UserIdentityConfiguration
from azure.identity import InteractiveBrowserCredential
from azure.keyvault.secrets import SecretClient


PROJECT_ROOT = Path(__file__).parent.resolve()
COMPUTE_NAME = "otel-cpu-cluster"
SECRET_NAME = "new-relic-license-key"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--subscription-id", required=True)
    parser.add_argument("--resource-group", required=True)
    parser.add_argument("--workspace-name", required=True)
    parser.add_argument("--otlp-endpoint", required=True)
    return parser.parse_args()

What does this foundation do?

• The imports provide Azure Machine Learning job definitions, browser authentication, Key Vault access, hidden input, and UUID generation.
• The constants keep the compute name and secret name consistent across every submission.
• The argument parser collects the Azure workspace location plus the New Relic endpoint.

• Add the initial main() function below parse_args():

def main() -> None:
    args = parse_args()
    license_key = getpass.getpass("Paste the New Relic ingest license key: ")
    if not license_key:
        raise ValueError("The New Relic ingest license key cannot be empty")

    credential = InteractiveBrowserCredential()
    try:
        ml_client = MLClient(
            credential=credential,
            subscription_id=args.subscription_id,
            resource_group_name=args.resource_group,
            workspace_name=args.workspace_name,
        )
        workspace = ml_client.workspaces.get(args.workspace_name)
        vault_name = workspace.key_vault.split("/")[-1]
        key_vault_url = f"https://{vault_name}.vault.azure.net/"

        with SecretClient(
            vault_url=key_vault_url,
            credential=credential,
        ) as secret_client:
            secret_client.set_secret(SECRET_NAME, license_key)
            stored_secret = secret_client.get_secret(SECRET_NAME)
            print(f"Stored Key Vault secret: {stored_secret.name}")
    finally:
        credential.close()

How is the key protected?

• The hidden prompt keeps the license key out of the command and source file.
• Interactive browser authentication connects the script to your Azure identity.
• The set and get operations confirm that your identity can use the workspace-linked vault.
• The credential closes even when a permission check fails.

• Save submit_pipeline.py.
• Check the current script for syntax errors by running:

python -m py_compile submit_pipeline.py

What does this check prove?

The command compiles the file without submitting a job. You return to the prompt without an error when the current structure is valid.

Seeing a syntax error?

Check that main() lines up with parse_args(). Confirm that the code inside try: and finally: uses four spaces of indentation.

Help me fix the syntax in my submission script.

• Insert the compute setup immediately above finally: in submit_pipeline.py:

        compute_names = {resource.name for resource in ml_client.compute.list()}
        if COMPUTE_NAME not in compute_names:
            compute = AmlCompute(
                name=COMPUTE_NAME,
                size="STANDARD_DS3_v2",
                min_instances=0,
                max_instances=1,
                idle_time_before_scale_down=120,
            )
            ml_client.compute.begin_create_or_update(compute).result()
            print(f"Created compute: {COMPUTE_NAME}")
        else:
            print(f"Reusing compute: {COMPUTE_NAME}")

How does compute stay controlled?

• The script reuses otel-cpu-cluster when it already exists.
• A new cluster uses one STANDARD_DS3_v2 node at most.
• The minimum of 0 lets the cluster release its idle node.
• The idle setting starts scale-down after 120 seconds.

• Insert the remote environment definition immediately below the compute setup:

        runtime_environment = Environment(
            image="mcr.microsoft.com/azureml/openmpi5.0-ubuntu24.04",
            conda_file=PROJECT_ROOT / "environment.yml",
        )

What defines the remote runtime?

The Azure Machine Learning image supplies the base runtime. Your existing environment.yml adds Python plus the pinned Azure and OpenTelemetry packages.

• Save submit_pipeline.py.
• Compile the expanded script by running:

python -m py_compile submit_pipeline.py

What does this check prove?

A silent return confirms that the compute and environment definitions fit inside the existing function structure.

• Insert the component definition immediately below runtime_environment:

        process_component = command(
            name="observable_order_processor",
            display_name="Observable order processor",
            code=PROJECT_ROOT / "src",
            command=(
                "python process_data.py "
                "--input ${{inputs.input_csv}} "
                "--output-dir ${{outputs.summary}} "
                "--run-id ${{inputs.run_id}} "
                "--exporter-mode otlp "
                "--key-vault-url ${{inputs.key_vault_url}} "
                "--secret-name ${{inputs.secret_name}} "
                "--otlp-endpoint ${{inputs.otlp_endpoint}}"
            ),
            inputs={
                "input_csv": Input(type="uri_file"),
                "run_id": Input(type="string"),
                "key_vault_url": Input(type="string"),
                "secret_name": Input(type="string"),
                "otlp_endpoint": Input(type="string"),
            },
            outputs={"summary": Output(type="uri_folder", mode="rw_mount")},
            environment=runtime_environment,
            identity=UserIdentityConfiguration(),
            is_deterministic=False,
        )

What does the component definition do?

• The command runs your existing process_data.py file in otlp mode.
• The run ID, vault URL, secret name, and endpoint travel as ordinary component inputs.
• The license key stays in Key Vault.
• User identity passthrough lets the job retrieve that key at runtime.

• Insert the pipeline definition immediately below process_component:

        correlation_id = str(uuid.uuid4())

        @dsl.pipeline(
            compute=COMPUTE_NAME,
            description="Observable order-processing proof of concept",
        )
        def observable_pipeline():
            step = process_component(
                input_csv=Input(
                    type="uri_file",
                    path=str(PROJECT_ROOT / "data" / "orders_bad.csv"),
                ),
                run_id=correlation_id,
                key_vault_url=key_vault_url,
                secret_name=SECRET_NAME,
                otlp_endpoint=args.otlp_endpoint,
            )
            return {"summary": step.outputs.summary}

How does the pipeline preserve correlation?

A new UUID becomes the component's run ID. The same value reaches every structured log and metric through your telemetry adapter.

The pipeline deliberately uses orders_bad.csv. The malformed row proves that the remote component records a failed record without failing the whole job.

• Save submit_pipeline.py.
• Compile the component and pipeline definitions by running:

python -m py_compile submit_pipeline.py

What does this check prove?

A silent return confirms that the component call and nested pipeline function are syntactically valid.

• Insert the submission logic immediately below observable_pipeline():

        pipeline_job = observable_pipeline()
        pipeline_job.display_name = "OpenTelemetry Azure ML pipeline"
        pipeline_job.tags = {"correlation_id": correlation_id}
        submitted_job = ml_client.jobs.create_or_update(
            pipeline_job,
            experiment_name="otel-azureml-observability",
        )

        print(f"Correlation ID: {correlation_id}")
        print(f"Studio URL: {submitted_job.studio_url}")
        ml_client.jobs.stream(submitted_job.name)
        final_job = ml_client.jobs.get(submitted_job.name)
        print(f"Pipeline finished with status: {final_job.status}")

What happens after submission?

The script tags the job with the correlation ID. It prints the identifier before streaming the remote logs.

The final lookup reads the completed job state from Azure Machine Learning. That state becomes your end-of-step proof.

• Add the script entry point below the completed main() function:

if __name__ == "__main__":
    main()

Why add an entry point?

The entry point starts main() when you run the file from the terminal. Importing the file from another module does not submit a pipeline.

• Save submit_pipeline.py.
• Compile the complete submission script by running:

python -m py_compile submit_pipeline.py

What does this check prove?

A silent return confirms that the complete submission script can be compiled before it touches Azure resources.

Submission script still not compiling?

Confirm that every inserted block remains inside try:. The finally: line must stay aligned with try:.

Help me compare my complete submission script.

Use these tabs to compare your completed submission file.

✔️ Awesome, I've got everything!

Your submission script is saved and compiles without an error.

ⓧ I'd like to double check the full code

from __future__ import annotations

import argparse
import getpass
import uuid
from pathlib import Path

from azure.ai.ml import Input, MLClient, Output, command, dsl
from azure.ai.ml.entities import AmlCompute, Environment, UserIdentityConfiguration
from azure.identity import InteractiveBrowserCredential
from azure.keyvault.secrets import SecretClient


PROJECT_ROOT = Path(__file__).parent.resolve()
COMPUTE_NAME = "otel-cpu-cluster"
SECRET_NAME = "new-relic-license-key"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--subscription-id", required=True)
    parser.add_argument("--resource-group", required=True)
    parser.add_argument("--workspace-name", required=True)
    parser.add_argument("--otlp-endpoint", required=True)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    license_key = getpass.getpass("Paste the New Relic ingest license key: ")
    if not license_key:
        raise ValueError("The New Relic ingest license key cannot be empty")

    credential = InteractiveBrowserCredential()
    try:
        ml_client = MLClient(
            credential=credential,
            subscription_id=args.subscription_id,
            resource_group_name=args.resource_group,
            workspace_name=args.workspace_name,
        )
        workspace = ml_client.workspaces.get(args.workspace_name)
        vault_name = workspace.key_vault.split("/")[-1]
        key_vault_url = f"https://{vault_name}.vault.azure.net/"

        with SecretClient(
            vault_url=key_vault_url,
            credential=credential,
        ) as secret_client:
            secret_client.set_secret(SECRET_NAME, license_key)
            stored_secret = secret_client.get_secret(SECRET_NAME)
            print(f"Stored Key Vault secret: {stored_secret.name}")

        compute_names = {resource.name for resource in ml_client.compute.list()}
        if COMPUTE_NAME not in compute_names:
            compute = AmlCompute(
                name=COMPUTE_NAME,
                size="STANDARD_DS3_v2",
                min_instances=0,
                max_instances=1,
                idle_time_before_scale_down=120,
            )
            ml_client.compute.begin_create_or_update(compute).result()
            print(f"Created compute: {COMPUTE_NAME}")
        else:
            print(f"Reusing compute: {COMPUTE_NAME}")

        runtime_environment = Environment(
            image="mcr.microsoft.com/azureml/openmpi5.0-ubuntu24.04",
            conda_file=PROJECT_ROOT / "environment.yml",
        )
        process_component = command(
            name="observable_order_processor",
            display_name="Observable order processor",
            code=PROJECT_ROOT / "src",
            command=(
                "python process_data.py "
                "--input ${{inputs.input_csv}} "
                "--output-dir ${{outputs.summary}} "
                "--run-id ${{inputs.run_id}} "
                "--exporter-mode otlp "
                "--key-vault-url ${{inputs.key_vault_url}} "
                "--secret-name ${{inputs.secret_name}} "
                "--otlp-endpoint ${{inputs.otlp_endpoint}}"
            ),
            inputs={
                "input_csv": Input(type="uri_file"),
                "run_id": Input(type="string"),
                "key_vault_url": Input(type="string"),
                "secret_name": Input(type="string"),
                "otlp_endpoint": Input(type="string"),
            },
            outputs={"summary": Output(type="uri_folder", mode="rw_mount")},
            environment=runtime_environment,
            identity=UserIdentityConfiguration(),
            is_deterministic=False,
        )

        correlation_id = str(uuid.uuid4())

        @dsl.pipeline(
            compute=COMPUTE_NAME,
            description="Observable order-processing proof of concept",
        )
        def observable_pipeline():
            step = process_component(
                input_csv=Input(
                    type="uri_file",
                    path=str(PROJECT_ROOT / "data" / "orders_bad.csv"),
                ),
                run_id=correlation_id,
                key_vault_url=key_vault_url,
                secret_name=SECRET_NAME,
                otlp_endpoint=args.otlp_endpoint,
            )
            return {"summary": step.outputs.summary}

        pipeline_job = observable_pipeline()
        pipeline_job.display_name = "OpenTelemetry Azure ML pipeline"
        pipeline_job.tags = {"correlation_id": correlation_id}
        submitted_job = ml_client.jobs.create_or_update(
            pipeline_job,
            experiment_name="otel-azureml-observability",
        )

        print(f"Correlation ID: {correlation_id}")
        print(f"Studio URL: {submitted_job.studio_url}")
        ml_client.jobs.stream(submitted_job.name)
        final_job = ml_client.jobs.get(submitted_job.name)
        print(f"Pipeline finished with status: {final_job.status}")
    finally:
        credential.close()


if __name__ == "__main__":
    main()

The cleanup script gives you a repeatable teardown path for the resources created by the submission.

• Create cleanup.py beside submit_pipeline.py by using your editor's file sidebar.
• Paste the cleanup foundation into cleanup.py:

from __future__ import annotations

import argparse

from azure.ai.ml import MLClient
from azure.identity import InteractiveBrowserCredential
from azure.keyvault.secrets import SecretClient


COMPUTE_NAME = "otel-cpu-cluster"
SECRET_NAME = "new-relic-license-key"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--subscription-id", required=True)
    parser.add_argument("--resource-group", required=True)
    parser.add_argument("--workspace-name", required=True)
    return parser.parse_args()

What does the cleanup foundation define?

The cleanup script targets the same compute cluster and Key Vault secret as the submission script. Its arguments identify the workspace that owns those resources.

• Add the initial cleanup function below parse_args():

def main() -> None:
    args = parse_args()
    credential = InteractiveBrowserCredential()
    try:
        ml_client = MLClient(
            credential=credential,
            subscription_id=args.subscription_id,
            resource_group_name=args.resource_group,
            workspace_name=args.workspace_name,
        )
        workspace = ml_client.workspaces.get(args.workspace_name)
        vault_name = workspace.key_vault.split("/")[-1]
        key_vault_url = f"https://{vault_name}.vault.azure.net/"
    finally:
        credential.close()


if __name__ == "__main__":
    main()

How does cleanup find the resources?

The script signs in with the same browser credential. It derives the workspace-linked vault URL from the selected Azure Machine Learning workspace.

• Save cleanup.py.
• Compile the cleanup foundation by running:

python -m py_compile cleanup.py

What does this check prove?

A silent return confirms that the cleanup foundation has valid Python syntax.

• Insert the compute deletion block immediately above finally: in cleanup.py:

        compute_names = {resource.name for resource in ml_client.compute.list()}
        if COMPUTE_NAME in compute_names:
            ml_client.compute.begin_delete(COMPUTE_NAME).result()
            print(f"Deleted compute: {COMPUTE_NAME}")
        else:
            print(f"Compute not found: {COMPUTE_NAME}")

How does compute cleanup stay repeatable?

The script checks whether otel-cpu-cluster exists before requesting deletion. Running cleanup again reports that the compute is absent.

• Insert the secret deletion block immediately below the compute cleanup:

        try:
            with SecretClient(
                vault_url=key_vault_url,
                credential=credential,
            ) as secret_client:
                secret_client.begin_delete_secret(SECRET_NAME).result()
            print(f"Began deleting Key Vault secret: {SECRET_NAME}")
        except Exception as error:
            print(f"Secret cleanup needs review: {error}")

How is secret deletion handled?

The Key Vault client begins deleting new-relic-license-key. A cleanup problem is printed for review without hiding the compute deletion result.

• Save cleanup.py.
• Compile the completed cleanup script by running:

python -m py_compile cleanup.py

What does this check prove?

A silent return confirms that the teardown logic is ready for the cleanup section. Do not run the cleanup script while you still need the remote resources.

Cleanup script not compiling?

Confirm that both deletion blocks sit inside the outer try:. Keep the secret cleanup's nested try: aligned with the compute check.

Help me fix my cleanup script.

Use these tabs to compare your completed cleanup file.

✔️ Awesome, I've got everything!

Your cleanup script is saved and ready for the teardown section.

ⓧ I'd like to double check the full code

from __future__ import annotations

import argparse

from azure.ai.ml import MLClient
from azure.identity import InteractiveBrowserCredential
from azure.keyvault.secrets import SecretClient


COMPUTE_NAME = "otel-cpu-cluster"
SECRET_NAME = "new-relic-license-key"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--subscription-id", required=True)
    parser.add_argument("--resource-group", required=True)
    parser.add_argument("--workspace-name", required=True)
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    credential = InteractiveBrowserCredential()
    try:
        ml_client = MLClient(
            credential=credential,
            subscription_id=args.subscription_id,
            resource_group_name=args.resource_group,
            workspace_name=args.workspace_name,
        )
        workspace = ml_client.workspaces.get(args.workspace_name)
        vault_name = workspace.key_vault.split("/")[-1]
        key_vault_url = f"https://{vault_name}.vault.azure.net/"

        compute_names = {resource.name for resource in ml_client.compute.list()}
        if COMPUTE_NAME in compute_names:
            ml_client.compute.begin_delete(COMPUTE_NAME).result()
            print(f"Deleted compute: {COMPUTE_NAME}")
        else:
            print(f"Compute not found: {COMPUTE_NAME}")

        try:
            with SecretClient(
                vault_url=key_vault_url,
                credential=credential,
            ) as secret_client:
                secret_client.begin_delete_secret(SECRET_NAME).result()
            print(f"Began deleting Key Vault secret: {SECRET_NAME}")
        except Exception as error:
            print(f"Secret cleanup needs review: {error}")
    finally:
        credential.close()


if __name__ == "__main__":
    main()

Submit with Key Vault protection

The first remote submission also tests your Key Vault data permissions. A successful set and get proves that the job can rely on the stored secret.

Your license key stays hidden

Entering a credential can feel risky. The script uses a hidden terminal prompt before storing the value in Key Vault.

The license key never enters the source file or job metadata. Avoid taking a screenshot while typing it.

Compute can incur charges

A STANDARD_DS3_v2 node incurs Azure VM charges while allocated. This project limits the cluster to one node.

The cluster scales to zero after it becomes idle. Associated Azure resources can still incur charges until cleanup.

• Choose the command for your New Relic data region below.
• Replace your-subscription-id with your Azure subscription ID.
• Replace your-resource-group with the workspace resource group.
• Replace your-workspace-name with the Azure Machine Learning workspace name.
• Allow the browser sign-in when it opens.
• Paste the New Relic ingest license key into the hidden terminal prompt.

Before you run the submission, which outcome do you expect from the malformed order on remote compute?

US data region

• Submit the pipeline to the US New Relic OTLP endpoint by running:

python submit_pipeline.py --subscription-id your-subscription-id --resource-group your-resource-group --workspace-name your-workspace-name --otlp-endpoint https://otlp.nr-data.net

What does this command do?

The command targets your Azure Machine Learning workspace. It sends remote telemetry to the US New Relic OTLP endpoint.

EU data region

• Submit the pipeline to the EU New Relic OTLP endpoint by running:

python submit_pipeline.py --subscription-id your-subscription-id --resource-group your-resource-group --workspace-name your-workspace-name --otlp-endpoint https://otlp.eu01.nr-data.net

What does this command do?

The command targets your Azure Machine Learning workspace. It sends remote telemetry to the EU New Relic OTLP endpoint.

The browser sign-in returns control to the terminal. The next output reveals whether your identity can write and read the workspace secret.

✔️ Pipeline completed

You first see a line beginning with Stored Key Vault secret:. This confirms secret get and set permissions.

The remote run can take several minutes while compute and the environment become ready. Keep the terminal open while the job stream reports progress.

That is the remote path working. Your job retrieves the secret through its delegated identity before exporting telemetry.

ⓧ I see Forbidden

The permission check has found a Key Vault data-plane gap. Your license key was not added to the job configuration.

• Ask an Azure administrator for the Key Vault Secrets Officer role at the workspace vault scope when the vault uses Azure RBAC.
• Ask for equivalent secret get and set permissions when the vault uses access policies.
• Return to your selected endpoint tab after the permission is granted.
• Rerun the same submission command.

Remote run not completing?

Check that the subscription ID, resource group, and workspace name point to the same workspace. Confirm that your Azure permissions allow compute creation.

Regional VM quota can also block a new cluster. Ask your Azure administrator whether STANDARD_DS3_v2 capacity is available for the workspace.

Help me diagnose my Azure ML submission.

Confirm the completed remote run

The malformed A103 row now tests the behavior you built locally. Before the stream finishes, do you expect that row to fail the whole pipeline?

• Wait for the streamed run to reach its final status.
• Confirm that the terminal contains a line beginning with Correlation ID:.
• Record the printed value as your correlation ID.
• Open the printed Studio URL: link in your browser.
• Confirm that the pipeline job shows Completed in Azure Machine Learning studio.
• Return to the terminal stream from earlier.
• Confirm that the final line reads Pipeline finished with status: Completed.

The remote pipeline completes even though the input contains not-a-number. Your observed processor counts that row as a failed record while preserving the three valid records.

Excellent work. Your component now runs on Azure Machine Learning with a Key Vault protected credential and a searchable correlation ID.

Your remote pipeline has exported its structured logs and metrics. Next, you will use your correlation ID to locate that exact run in New Relic.

Correlate the Remote Run

Your Azure Machine Learning pipeline now completes with a correlation ID. That identifier is the thread connecting the remote job to its telemetry.

A successful job status proves that the component ran. An operator still needs to locate its health signals plus the exact failed record.

You will use NRQL in New Relic to find one run across logs plus metrics. The shared pipeline.run.id makes that correlation possible.

In this step, get ready to:

• Facet recent logs by pipeline.run.id plus severity.text.
• Query the processed counter plus failed counter plus duration histogram by pipeline.run.id.
• Trace malformed order A103 from aggregate metrics to its structured failure event.

Query the remote logs

OpenTelemetry resource attributes travel with each exported log record. Grouping by pipeline.run.id separates this pipeline run from other telemetry using the same service name.

• Sign in to your New Relic account in your browser.
• Open Query your data.

Before you run the first query, do you expect the completed pipeline to appear under one run identifier or several?

• Count the remote log records from the previous 30 minutes by running this query:

FROM Log SELECT count(*)
WHERE service.name = 'azureml-order-processor'
FACET pipeline.run.id, severity.text
SINCE 30 minutes ago

What does this query do?

• The Log data type contains the exported log records.
• The count(*) function counts matching records.
• The service.name filter limits the result to the order processor.
• The FACET clause groups records by run identifier plus severity.
• The SINCE 30 minutes ago clause keeps the search focused on the remote run you just completed.

• Match the pipeline.run.id facet to the correlation ID from the completed pipeline output.

You should see the matching run separated into severity facets. This proves that the printed correlation ID reached New Relic with the remote logs.

No matching log facet?

• Confirm that the query still includes SINCE 30 minutes ago.
• Confirm that your selected New Relic account matches the account tied to the ingest license key.
• Repeat the pipeline submission from the previous step if the original job completed outside the query window.

Help me diagnose missing logs from this Azure Machine Learning pipeline.

Query the remote metrics

The metric view summarizes the workload behind that run identifier. Its counters reveal record outcomes while its histogram reports component duration.

Before you run this query, what totals do you expect from the three valid rows plus the malformed row?

• Aggregate the remote metrics from the previous 30 minutes by running this query:

FROM Metric SELECT sum(pipeline.records.processed),
  sum(pipeline.records.failed),
  average(pipeline.component.duration)
WHERE service.name = 'azureml-order-processor'
FACET pipeline.run.id
SINCE 30 minutes ago

What does this query measure?

• The Metric data type contains the exported counters plus histogram.
• The sum(pipeline.records.processed) expression totals valid records.
• The sum(pipeline.records.failed) expression totals rejected records.
• The average(pipeline.component.duration) expression returns the recorded execution duration.
• The pipeline.run.id facet keeps each pipeline execution separate.

• Locate the metric facet matching the correlation ID from submit_pipeline.py.

You should see 3 processed records. You should also see 1 failed record.

The same facet should contain a value for pipeline.component.duration. That value confirms the histogram reached New Relic before the batch process exited.

Logs appear but metrics are missing?

• Confirm that the query reads from Metric.
• Confirm that the query filters on service.name = 'azureml-order-processor'.
• Rerun the metric query after refreshing the query results.

Help me diagnose missing OpenTelemetry metrics in New Relic.

Inspect the failed record

Aggregate metrics tell you that one record failed. The structured log identifies the business record plus the row that caused that failure.

• Return to the log query from the first substep.
• Open the log events behind the matching pipeline.run.id facet.
• Select the event named record_failed.
• Inspect the event attributes.

You should see order.id set to A103. The record.row attribute should identify row 5.

The exception attributes should describe the failed conversion of not-a-number. This is the searchable failure context that the fail-fast baseline lacked.

Before you compare everything, which shared field should connect the aggregate metrics to this exact failed log?

• Match the correlation ID from submit_pipeline.py to pipeline.run.id in both result sets.
• Confirm that the metric facet shows 3 processed records.
• Confirm that the metric facet shows 1 failed record.
• Confirm that the metric facet contains a pipeline.component.duration value.
• Confirm that the record_failed event identifies order A103.

Failure attributes missing?

• Confirm that you selected the record_failed event from the matching run facet.
• Confirm that the selected event belongs to service azureml-order-processor.
• Expand the event attributes before checking order.id plus record.row.

Help me find the failed record attributes in New Relic.

That is the operator workflow working from end to end. You can now move from one pipeline identifier to aggregate health plus the exact failed record.

Secret mission

Turn Failures into an Alert

Your failed-record metric already exposes a broken row. Turn that signal into a static New Relic alert condition that is ready to notify operators when a batch reports failures.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep the project available, let its compute return to zero nodes, or remove its teardownable resources. Azure VM charges apply while a node is allocated.

Cost warning

The otel-cpu-cluster avoids idle node compute charges after it scales to zero. Associated Azure resources can still incur costs until deletion.

New Relic ingestion follows your existing account plan. Deleting the project resources gives you the strictest Azure cleanup.

Resources you used:

• Local project files, including requirements.txt, environment.yml, data, src, outputs, submit_pipeline.py, and cleanup.py.
• Azure Machine Learning compute cluster otel-cpu-cluster.
• Azure Key Vault secret new-relic-license-key.
• New Relic alert condition named P2 | Failed records | Azure ML order processor.

What Is Outside This Cleanup?

Azure Machine Learning job history remains available after cleanup.py runs. Its code snapshot remains available too.

The environment asset stays in Azure Machine Learning. The remote output artifact and ingested New Relic telemetry also remain available.

Keep everything running

No action is needed. Choose this option if you plan to reuse the observable component for another pipeline run.

• Your local source files remain available for further development.
• The otel-cpu-cluster remains available. It returns to zero nodes while idle.
• The new-relic-license-key secret remains in the workspace Key Vault.
• The P2 | Failed records | Azure ML order processor alert condition keeps evaluating future telemetry.
• Monitor Azure VM allocation before each new run.
• Monitor New Relic ingestion through your existing account plan.

Pause - I'll come back to this later

Let remote compute return to zero nodes while keeping the project ready for later. This stops active node usage without removing your project resources.

• Wait at least 120 seconds after the completed job for the idle scale-down window.
• Return to the cluster view in your Azure Machine Learning workspace.
• Confirm that otel-cpu-cluster shows zero active nodes.
• Keep the local project files for your next session.
• Keep the new-relic-license-key secret for future pipeline exports.
• Keep the New Relic alert condition for future failed-record signals.
• Monitor associated Azure resource costs while the cluster exists.

Delete - I don't want to use this again

Deletion permanently removes the teardownable project resources. The Azure cleanup script removes the compute cluster first.

The script then begins deleting the Key Vault secret. Azure may request authentication in your browser.

• Confirm that you no longer need the teardownable Azure resources.
• Use a terminal in the folder that contains cleanup.py.
• Replace your-subscription-id in the command below with the subscription ID used for the pipeline.
• Replace your-resource-group with the resource group used for the pipeline.
• Replace your-workspace-name with the Azure Machine Learning workspace used for the pipeline.
• Delete the Azure resources by running this command:

python cleanup.py --subscription-id your-subscription-id --resource-group your-resource-group --workspace-name your-workspace-name

What Does This Cleanup Do?

• The script deletes otel-cpu-cluster when the compute cluster exists.
• The script begins deleting new-relic-license-key from the workspace Key Vault.

• Complete Azure authentication if your browser requests it.
• Wait for the cleanup command to finish.

You should see Deleted compute: otel-cpu-cluster followed by Began deleting Key Vault secret: new-relic-license-key. The highest-risk Azure cost source is now removed.

Did the Azure Cleanup Stop Early?

Confirm that your account can delete Azure Machine Learning compute. Confirm that it can delete secrets from the workspace Key Vault.

Help me diagnose the Azure cleanup.

The New Relic alert condition sits outside the Azure cleanup. It must be removed from the policy where you created it.

• Return to All capabilities in New Relic.
• Select Alerts.
• Select Alert conditions (Policies).
• Open the policy that contains P2 | Failed records | Azure ML order processor.
• Locate the P2 | Failed records | Azure ML order processor condition.
• Use the condition's deletion control to remove it.

The condition should no longer appear in the selected policy. Future failed-record metrics no longer trigger this project alert.

Can't Find the Alert Condition?

Confirm that you opened the same policy used during the Secret Mission. Search for the full condition name inside that policy.

Help me find the New Relic alert condition.

The local project files are the final teardown target. Choose your platform below.

macOS

This command permanently removes the local source files and generated summaries. The surrounding folder remains because its name was not fixed by the project.

• Confirm that the terminal is still in the folder containing cleanup.py.
• Delete the local project files by running this command:

rm -rf data src outputs requirements.txt environment.yml submit_pipeline.py cleanup.py

What Does This Command Remove?

The command removes the sample data and Python source files. It also removes the local summaries and project configuration files.

The submission script and cleanup script are removed in the same operation. The unnamed containing folder remains on your computer.

• Refresh your editor's file explorer.

You should no longer see data, src, outputs, or the four project files.

Are Local Files Still Present?

Confirm that the terminal was in the folder containing cleanup.py before you ran the command. Check whether your account has permission to delete the remaining files.

Help me remove the remaining local files.

Windows

These commands use Windows Command Prompt. They permanently remove the local source files and generated summaries.

• Switch the shell in your editor's terminal panel to Windows Command Prompt.
• Confirm that the prompt is in the folder containing cleanup.py.
• Delete the local project files by running these commands:

rmdir /s /q data
rmdir /s /q src
rmdir /s /q outputs
del /q requirements.txt
del /q environment.yml
del /q submit_pipeline.py
del /q cleanup.py

What Do These Commands Remove?

The rmdir commands remove the sample data and Python source folders. They also remove the folder containing local summary outputs.

The del commands remove the project configuration files and Python scripts. The unnamed containing folder remains on your computer.

• Refresh your editor's file explorer.

You should no longer see data, src, outputs, or the four project files.

Are Local Files Still Present?

Confirm that the terminal used Windows Command Prompt. Confirm that the prompt was in the folder containing cleanup.py.

Help me remove the remaining local files.

Linux

This command permanently removes the local source files and generated summaries. The surrounding folder remains because its name was not fixed by the project.

• Confirm that the terminal is still in the folder containing cleanup.py.
• Delete the local project files by running this command:

rm -rf data src outputs requirements.txt environment.yml submit_pipeline.py cleanup.py

What Does This Command Remove?

The command removes the sample data and Python source files. It also removes the local summaries and project configuration files.

The submission script and cleanup script are removed in the same operation. The unnamed containing folder remains on your computer.

• Refresh your editor's file explorer.

You should no longer see data, src, outputs, or the four project files.

Are Local Files Still Present?

Confirm that the terminal was in the folder containing cleanup.py before you ran the command. Check whether your account has permission to delete the remaining files.

Help me remove the remaining local files.

Nice Work!

Nice Work!

You made it! You built an observable Azure Machine Learning batch pipeline with secure OpenTelemetry export to New Relic.

What you learned:

• Captured malformed order A103 with structured failure context. The remaining three records still produced a summary.
• Added record counters that separate processed rows from failed rows. A duration histogram now shows how long each component run takes.
• Ran an identity-aware remote pipeline on Azure Machine Learning compute. Azure Key Vault protected the license key. One correlation ID connected the exact run to its New Relic logs and metrics.
• Secret Mission: Created the static NRQL alert condition P2 | Failed records | Azure ML order processor. It surfaces failed batch records before manual investigation.

Ready to quiz yourself?