How to Use AWS Macie 2026 and OpenTelemetry 1.20 to Detect PII Leaks in Logs

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/macie2"
    "github.com/aws/aws-sdk-go-v2/service/macie2/types"
    "github.com/aws/aws-sdk-go-v2/service/s3"
)

const (
    macieRegion     = "us-east-1"
    logBucketName   = "prod-app-logs-2026"
    classifierJobID = "daily-log-pii-scan"
)

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // Load AWS config with shared credentials or IAM role
    cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(macieRegion))
    if err != nil {
        log.Fatalf("failed to load AWS config: %v", err)
    }

    // Initialize Macie v2 client (2026 uses same API version as 2023, but adds new detector types)
    macieClient := macie2.NewFromConfig(cfg)
    s3Client := s3.NewFromConfig(cfg)

    // Verify target log bucket exists and is accessible
    _, err = s3Client.HeadBucket(ctx, &s3.HeadBucketInput{
        Bucket: aws.String(logBucketName),
    })
    if err != nil {
        log.Fatalf("log bucket %s not accessible: %v", logBucketName, err)
    }
    fmt.Printf("Verified log bucket %s exists
", logBucketName)

    // Enable Macie for the account if not already enabled
    _, err = macieClient.EnableMacie(ctx, &macie2.EnableMacieInput{
        FindingPublishingFrequency: types.FindingPublishingFrequencyFifteenMinutes,
        Status:                     types.MacieStatusEnabled,
    })
    if err != nil {
        // Ignore error if Macie is already enabled
        var alreadyEnabled *types.ConflictException
        if !errors.As(err, &alreadyEnabled) {
            log.Fatalf("failed to enable Macie: %v", err)
        }
        fmt.Println("Macie already enabled for account")
    } else {
        fmt.Println("Macie enabled successfully")
    }

    // Create custom classification job for log bucket
    // Macie 2026 adds native support for JSON, plain text, and NDJSON log formats
    jobOutput, err := macieClient.CreateClassificationJob(ctx, &macie2.CreateClassificationJobInput{
        ClientToken:       aws.String(classifierJobID),
        CustomDataIdentifierIds: nil, // Use managed PII detectors only
        Description:       aws.String("Daily scan of production app logs for PII leaks"),
        InitialStatus:     types.JobStatusRunning,
        JobType:           types.JobTypeScheduled,
        Name:              aws.String(classifierJobID),
        ScheduleFrequency: types.ScheduleFrequencyDaily,
        // Scan all NDJSON and plain text logs in the bucket
        S3JobDefinition: &types.S3JobDefinition{
            BucketDefinitions: []types.S3BucketDefinitionForJob{
                {
                    AccountId: aws.String(os.Getenv("AWS_ACCOUNT_ID")),
                    BucketName: aws.String(logBucketName),
                    // Only scan logs with .log or .ndjson suffix
                    ObjectKeyRegex: aws.String(".*\\.(log|ndjson)$"),
                },
            },
            // Use Macie 2026's managed PII detectors: email, SSN, credit card, phone number
            Scoping: &types.Scoping{
                Includes: []types.S3BucketCriteriaForJob{
                    {
                        // Only scan logs modified in last 7 days to reduce cost
                        LastModified: &types.DateRange{
                            End:   aws.Time(time.Now()),
                            Start: aws.Time(time.Now().Add(-7 * 24 * time.Hour)),
                        },
                    },
                },
            },
        },
    })
    if err != nil {
        log.Fatalf("failed to create classification job: %v", err)
    }
    fmt.Printf("Created Macie classification job: %s
", *jobOutput.JobId)
    fmt.Printf("View job in console: https://%s.console.aws.amazon.com/macie/home?region=%s#classification-jobs
", macieRegion, macieRegion)
}

package main

import (
    "context"
    "crypto/rand"
    "fmt"
    "log"
    "math/big"
    "os"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlplogs"
    "go.opentelemetry.io/otel/exporters/otlp/otlplogs/otlplogsgrpc"
    "go.opentelemetry.io/otel/log"
    "go.opentelemetry.io/otel/log/global"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/log"
    "go.opentelemetry.io/otel/sdk/resource"
    semconv "go.opentelemetry.io/otel/semconv/v1.20.0"
)

const (
    otelCollectorAddr = "otel-collector:4317" // gRPC port for OTel 1.20
    serviceName       = "pii-log-generator"
    piiLogCount       = 1000 // Number of synthetic PII logs to send
)

// generateSSN generates a fake SSN matching regex ^\d{3}-\d{2}-\d{4}$
func generateSSN() string {
    part1, _ := rand.Int(rand.Reader, big.NewInt(999))
    part2, _ := rand.Int(rand.Reader, big.NewInt(99))
    part3, _ := rand.Int(rand.Reader, big.NewInt(9999))
    return fmt.Sprintf("%03d-%02d-%04d", part1, part2, part3)
}

// generateEmail generates a fake email address
func generateEmail() string {
    domains := []string{"example.com", "test.org", "demo.net"}
    idx, _ := rand.Int(rand.Reader, big.NewInt(int64(len(domains))))
    return fmt.Sprintf("user%d@%s", time.Now().UnixNano(), domains[idx])
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    defer cancel()

    // Configure OTel 1.20 resource with service name and version
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceName(serviceName),
            semconv.ServiceVersion("1.0.0"),
        ),
    )
    if err != nil {
        log.Fatalf("failed to create OTel resource: %v", err)
    }

    // Initialize OTLP gRPC log exporter for OTel 1.20
    exporter, err := otlplogs.New(ctx, otlplogsgrpc.NewClient(
        otlplogsgrpc.WithInsecure(), // Use TLS in production
        otlplogsgrpc.WithEndpoint(otelCollectorAddr),
    ))
    if err != nil {
        log.Fatalf("failed to create OTLP log exporter: %v", err)
    }
    defer exporter.Shutdown(ctx)

    // Create log provider with batch processor (OTel 1.20 default batch size is 512 logs)
    provider := log.NewLoggerProvider(
        log.WithResource(res),
        log.WithProcessor(log.NewBatchProcessor(exporter)),
    )
    defer provider.Shutdown(ctx)
    global.SetLoggerProvider(provider)

    logger := provider.Logger(serviceName)

    fmt.Printf("Sending %d synthetic logs with PII to OTel collector at %s
", piiLogCount, otelCollectorAddr)

    for i := 0; i < piiLogCount; i++ {
        // Generate synthetic log with PII (SSN, email, credit card)
        logBody := fmt.Sprintf(
            `{"timestamp":"%s","level":"info","message":"User login","user_id":%d,"email":"%s","ssn":"%s","credit_card":"%s"}`,
            time.Now().Format(time.RFC3339),
            i,
            generateEmail(),
            generateSSN(),
            "4111-1111-1111-1111", // Fake Visa test card
        )

        // Create log record with severity and attributes
        record := log.Record{}
        record.SetBody(log.StringValue(logBody))
        record.SetSeverity(log.SeverityInfo)
        record.SetTimestamp(time.Now())
        record.AddAttributes(
            log.String("log.file", "auth.log"),
            log.String("env", "production"),
        )

        // Emit log to OTel collector
        logger.Emit(ctx, record)

        // Throttle to avoid overwhelming collector
        if i%100 == 0 {
            time.Sleep(10 * time.Millisecond)
        }
    }

    fmt.Printf("Successfully sent %d logs with PII
", piiLogCount)
}

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  logstransform:
    redaction:
      # Use OTel 1.20's built-in PII patterns
      patterns:
        - email
        - ssn
        - credit_card
        - phone_number
      # Replace PII with [REDACTED] instead of removing
      replacement: "[REDACTED]"
  batch:
    timeout: 5s
    send_batch_size: 512

exporters:
  s3:
    bucket: prod-app-logs-2026
    region: us-east-1
    prefix: redacted-logs/
  logging:
    verbosity: detailed

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [logstransform, batch]
      exporters: [s3, logging]

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "os"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/macie2"
    "github.com/aws/aws-sdk-go-v2/service/macie2/types"
    "github.com/slack-go/slack"
)

const (
    slackToken     = os.Getenv("SLACK_TOKEN")
    slackChannel   = "#security-alerts"
    macieRegion    = "us-east-1"
    listenAddr     = ":8080"
)

// MacieFinding represents a PII leak finding from Macie 2026
type MacieFinding struct {
    ID          string   `json:"id"`
    Type        string   `json:"type"`
    Severity    string   `json:"severity"`
    Description string   `json:"description"`
    Resources   []string `json:"resources"`
    AccountID   string   `json:"accountId"`
    Region      string   `json:"region"`
    CreatedAt   string   `json:"createdAt"`
}

func main() {
    ctx := context.Background()

    // Initialize Slack client
    if slackToken == "" {
        log.Fatal("SLACK_TOKEN environment variable is required")
    }
    slackClient := slack.New(slackToken)

    // Initialize Macie client to list findings
    cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(macieRegion))
    if err != nil {
        log.Fatalf("failed to load AWS config: %v", err)
    }
    macieClient := macie2.NewFromConfig(cfg)

    // Start HTTP server to receive Macie finding notifications (via EventBridge)
    http.HandleFunc("/macie-finding", func(w http.ResponseWriter, r *http.Request) {
        if r.Method != http.MethodPost {
            http.Error(w, "only POST allowed", http.StatusMethodNotAllowed)
            return
        }

        var finding MacieFinding
        if err := json.NewDecoder(r.Body).Decode(&finding); err != nil {
            log.Printf("failed to decode finding: %v", err)
            http.Error(w, "invalid request body", http.StatusBadRequest)
            return
        }

        // Map Macie severity to Slack color
        severityColor := "#36a64f" // Low
        switch finding.Severity {
        case "HIGH":
            severityColor = "#ff0000"
        case "MEDIUM":
            severityColor = "#ffa500"
        }

        // Send Slack alert
        _, _, err := slackClient.PostMessage(
            ctx,
            slackChannel,
            slack.MsgOptionAttachments(slack.Attachment{
                Color:     severityColor,
                Title:     fmt.Sprintf("Macie PII Leak Detected: %s", finding.Type),
                Text:      finding.Description,
                Fields: []slack.AttachmentField{
                    {Title: "Finding ID", Value: finding.ID, Short: true},
                    {Title: "Severity", Value: finding.Severity, Short: true},
                    {Title: "Resource", Value: fmt.Sprintf("%v", finding.Resources), Short: false},
                    {Title: "Account", Value: finding.AccountID, Short: true},
                    {Title: "Region", Value: finding.Region, Short: true},
                },
                Footer:     "AWS Macie 2026 + OpenTelemetry 1.20",
                FooterIcon: "https://platform.slack-edge.com/img/default_application_icon.png",
                Timestamp:  time.Now().Unix(),
            }),
        )
        if err != nil {
            log.Printf("failed to send Slack alert: %v", err)
            http.Error(w, "failed to send alert", http.StatusInternalServerError)
            return
        }

        log.Printf("Sent Slack alert for finding %s", finding.ID)
        w.WriteHeader(http.StatusOK)
    })

    // Also poll for Macie findings every 5 minutes as fallback
    go func() {
        ticker := time.NewTicker(5 * time.Minute)
        defer ticker.Stop()

        for range ticker.C {
            listOutput, err := macieClient.ListFindings(ctx, &macie2.ListFindingsInput{
                FindingCriteria: &types.FindingCriteria{
                    // Only get high and medium severity PII findings
                    Criterion: map[string]types.Criterion{
                        "severity": {
                            Eq: []string{"HIGH", "MEDIUM"},
                        },
                        "type": {
                            Eq: []string{"PII_LEAK"},
                        },
                    },
                },
                SortCriteria: &types.SortCriteria{
                    AttributeName: aws.String("createdAt"),
                    OrderBy:       types.OrderByDesc,
                },
            })
            if err != nil {
                log.Printf("failed to list Macie findings: %v", err)
                continue
            }

            for _, findingID := range listOutput.FindingIds {
                getOutput, err := macieClient.GetFinding(ctx, &macie2.GetFindingInput{
                    FindingId: aws.String(findingID),
                })
                if err != nil {
                    log.Printf("failed to get finding %s: %v", findingID, err)
                    continue
                }
                // Process finding (send to Slack if not already sent)
                log.Printf("Polled finding: %s, severity: %s", findingID, getOutput.Severity)
            }
        }
    }()

    fmt.Printf("Alert handler listening on %s
", listenAddr)
    log.Fatal(http.ListenAndServe(listenAddr, nil))
}

// List managed PII detectors in Macie 2026
listDetectorsOutput, err := macieClient.ListManagedDataIdentifiers(ctx, &macie2.ListManagedDataIdentifiersInput{
    Category: types.ManagedDataIdentifierCategoryPii,
})
if err != nil {
    log.Fatalf("failed to list managed detectors: %v", err)
}
for _, detector := range listDetectorsOutput.Items {
    fmt.Printf("Detector: %s, Description: %s
", *detector.Id, *detector.Description)
}

processors:
  sampling:
    rules:
      - name: sample-pii
        condition: body matches ".*@.*|\\d{3}-\\d{2}-\\d{4}"
        rate: 1.0 # Always sample logs with PII markers
      - name: sample-non-pii
        condition: body not matches ".*@.*|\\d{3}-\\d{2}-\\d{4}"
        rate: 0.1 # Sample 10% of non-PII logs

// Pseudocode for Jira integration
func createJiraTicket(finding MacieFinding) {
    jiraClient.CreateIssue(ctx, jira.Issue{
        Type: "Security Incident",
        Summary: fmt.Sprintf("PII Leak: %s", finding.Type),
        Description: finding.Description,
        Priority: mapSeverityToPriority(finding.Severity),
        Labels: []string{"pii", "macie"},
    })
}

pii-leak-detector/
├── cmd/
│   ├── setup-macie/
│   │   └── main.go
│   ├── send-logs/
│   │   └── main.go
│   └── alert-handler/
│       └── main.go
├── deploy/
│   ├── otel-collector/
│   │   ├── config.yaml
│   │   └── daemonset.yaml
│   └── k8s/
│       ├── macie-role.yaml
│       └── alert-handler-deployment.yaml
├── pkg/
│   ├── macie/
│   │   └── client.go
│   └── otel/
│       └── exporter.go
├── go.mod
├── go.sum
├── Makefile
└── README.md