otlp2pipeline 0.4.0

OTLP ingestion worker for Cloudflare Pipelines and AWS
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

use anyhow::Result;

use super::cli::AwsCli;
use super::context::S3_TABLES_ROLE_NAME;
use super::helpers::{load_config, resolve_env_with_config, resolve_region, stack_name};
use super::schema::TABLES;
use crate::cli::StatusArgs;

pub fn execute_status(args: StatusArgs) -> Result<()> {
    let config = load_config()?;
    let env_name = resolve_env_with_config(args.env, &config)?;
    let region = resolve_region(args.region, &config);
    let stack = stack_name(&env_name);

    let cli = AwsCli::new(&region);
    let account = cli.sts().get_caller_identity()?;

    eprintln!("Checking deployment status...\n");
    eprintln!("Account: {}", account.account_id);
    eprintln!("Region:  {}", region);
    eprintln!("Stack:   {}", stack);
    eprintln!();

    // S3 Tables Setup
    eprintln!("S3 Tables Setup:");

    // IAM Role
    if cli.iam().role_exists(S3_TABLES_ROLE_NAME)? {
        eprintln!("  [ok] IAM Role: {}", S3_TABLES_ROLE_NAME);
    } else {
        eprintln!("  [missing] IAM Role: {} (not found)", S3_TABLES_ROLE_NAME);
    }

    // LakeFormation resource
    let resource_arn = format!(
        "arn:aws:s3tables:{}:{}:bucket/*",
        region, account.account_id
    );
    if cli.lakeformation().describe_resource(&resource_arn)? {
        eprintln!("  [ok] LakeFormation Resource: registered");
    } else {
        eprintln!("  [missing] LakeFormation Resource: not registered");
    }

    // Glue catalog
    if cli.glue().catalog_exists("s3tablescatalog")? {
        eprintln!("  [ok] Glue Catalog: s3tablescatalog");
    } else {
        eprintln!("  [missing] Glue Catalog: s3tablescatalog (not found)");
    }

    eprintln!();

    // CloudFormation Stack
    eprintln!("CloudFormation Stack: {}", stack);
    let stack_info = cli.cloudformation().describe_stack(&stack)?;

    match stack_info {
        Some(info) => {
            eprintln!("  [ok] Status: {}", info.status);

            // Firehose streams
            eprintln!();
            eprintln!("Firehose Streams:");
            let mut all_ready = true;
            for table in TABLES {
                let stream_name = format!("{}-{}", stack, table);
                if cli.firehose().stream_exists(&stream_name)? {
                    eprintln!("  [ok] {} (AppendOnly: true)", stream_name);
                } else {
                    eprintln!("  [missing] {} (not found)", stream_name);
                    all_ready = false;
                }
            }

            // Lambda function
            eprintln!();
            eprintln!("Lambda Function:");
            let function_name = format!("{}-ingest", stack);
            if cli.lambda().function_exists(&function_name)? {
                eprintln!("  [ok] {}", function_name);
                if let Some(url) = cli.lambda().get_function_url(&function_name)? {
                    eprintln!();
                    eprintln!("OTLP Endpoints:");
                    eprintln!("  POST {}v1/logs", url);
                    eprintln!("  POST {}v1/traces", url);
                    eprintln!("  POST {}v1/metrics", url);
                }
            } else {
                eprintln!("  [missing] {} (not found)", function_name);
            }

            eprintln!();
            if all_ready {
                eprintln!("[ok] Deployment complete! Firehose is ready to receive data.");
            } else {
                eprintln!("[warn] Some Firehose streams are missing. Run deploy to create them.");
            }
        }
        None => {
            eprintln!("  [missing] Stack does not exist");
        }
    }

    Ok(())
}