Datadog入門 第4部 - アプリケーション監視の実践完全ガイド ​

// Go言語でのデータベースクエリトレーシング
package main

import (
    "context"
    "database/sql"
    "time"
    
    "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
    sqltrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/database/sql"
    
    _ "github.com/lib/pq"
)

func main() {
    // Datadogトレーサー初期化
    tracer.Start(
        tracer.WithService("product-service"),
        tracer.WithEnv("production"),
    )
    defer tracer.Stop()
    
    // データベース接続にトレーシングを追加
    sqltrace.Register("postgres", &pq.Driver{})
    db, err := sqltrace.Open("postgres", "postgres://localhost/mydb")
    if err != nil {
        panic(err)
    }
    defer db.Close()
}

func getProducts(ctx context.Context, db *sql.DB, categoryID int) ([]Product, error) {
    span, ctx := tracer.StartSpanFromContext(ctx, "database.query")
    defer span.Finish()
    
    span.SetTag("sql.query", "SELECT * FROM products WHERE category_id = $1")
    span.SetTag("db.instance", "products")
    
    start := time.Now()
    
    rows, err := db.QueryContext(ctx, `
        SELECT id, name, price, description 
        FROM products 
        WHERE category_id = $1 AND active = true
        ORDER BY popularity DESC
        LIMIT 100
    `, categoryID)
    
    duration := time.Since(start)
    span.SetTag("query.duration", duration.Milliseconds())
    
    if err != nil {
        span.SetTag("error", true)
        span.SetTag("error.message", err.Error())
        return nil, err
    }
    defer rows.Close()
    
    // 結果処理...
    var products []Product
    for rows.Next() {
        // スキャン処理
    }
    
    span.SetTag("query.rows_returned", len(products))
    return products, nil
}

// Go でのDatadog APM設定
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "strconv"
    "time"
    
    "github.com/gorilla/mux"
    "github.com/go-redis/redis/v8"
    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    
    "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
    "gopkg.in/DataDog/dd-trace-go.v1/contrib/gorilla/mux"
    "gopkg.in/DataDog/dd-trace-go.v1/contrib/go-redis/redis"
    "gopkg.in/DataDog/dd-trace-go.v1/contrib/jmoiron/sqlx"
)

type Server struct {
    db    *sqlx.DB
    redis *redis.Client
}

type Product struct {
    ID          int       `json:"id" db:"id"`
    Name        string    `json:"name" db:"name"`
    Price       float64   `json:"price" db:"price"`
    Category    string    `json:"category" db:"category"`
    Description string    `json:"description" db:"description"`
    CreatedAt   time.Time `json:"created_at" db:"created_at"`
}

func main() {
    // Datadogトレーサーの初期化
    tracer.Start(
        tracer.WithService("product-service"),
        tracer.WithEnv("production"),
        tracer.WithServiceVersion("1.0.0"),
        tracer.WithRuntimeMetrics(),
        tracer.WithProfilerCodeHotspots(true),
        tracer.WithProfilerEndpoints(true),
    )
    defer tracer.Stop()
    
    // データベース接続
    db, err := sqlx.Connect("postgres", "postgresql://localhost/products?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()
    
    // Redis接続
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    
    server := &Server{
        db:    db,
        redis: rdb,
    }
    
    // ルーターの設定
    r := mux.NewRouter()
    r.Use(muxtrace.Middleware(
        muxtrace.WithServiceName("product-service"),
    ))
    
    r.HandleFunc("/products", server.getProducts).Methods("GET")
    r.HandleFunc("/products/{id}", server.getProduct).Methods("GET")
    r.HandleFunc("/products", server.createProduct).Methods("POST")
    
    log.Println("Server starting on :8080")
    log.Fatal(http.ListenAndServe(":8080", r))
}

func (s *Server) getProducts(w http.ResponseWriter, r *http.Request) {
    span, ctx := tracer.StartSpanFromContext(r.Context(), "products.list")
    defer span.Finish()
    
    // クエリパラメータの取得
    category := r.URL.Query().Get("category")
    page, _ := strconv.Atoi(r.URL.Query().Get("page"))
    if page <= 0 {
        page = 1
    }
    limit, _ := strconv.Atoi(r.URL.Query().Get("limit"))
    if limit <= 0 || limit > 100 {
        limit = 20
    }
    
    span.SetTag("query.category", category)
    span.SetTag("query.page", page)
    span.SetTag("query.limit", limit)
    
    // キャッシュキーの生成
    cacheKey := fmt.Sprintf("products:list:%s:%d:%d", category, page, limit)
    
    // Redis キャッシュから取得を試行
    cacheSpan, ctx := tracer.StartSpanFromContext(ctx, "cache.get")
    var products []Product
    
    cachedData, err := s.redis.Get(ctx, cacheKey).Result()
    if err == nil {
        cacheSpan.SetTag("cache.hit", true)
        json.Unmarshal([]byte(cachedData), &products)
        span.SetTag("data.source", "cache")
    } else {
        cacheSpan.SetTag("cache.hit", false)
        span.SetTag("data.source", "database")
    }
    cacheSpan.Finish()
    
    // データベースから取得
    if len(products) == 0 {
        dbSpan, ctx := tracer.StartSpanFromContext(ctx, "database.query")
        
        query := `
            SELECT id, name, price, category, description, created_at
            FROM products
            WHERE ($1 = '' OR category = $1)
            ORDER BY created_at DESC
            LIMIT $2 OFFSET $3
        `
        
        offset := (page - 1) * limit
        
        dbSpan.SetTag("db.statement", query)
        dbSpan.SetTag("db.operation", "SELECT")
        dbSpan.SetTag("db.table", "products")
        
        err := s.db.SelectContext(ctx, &products, query, category, limit, offset)
        if err != nil {
            dbSpan.SetTag("error", true)
            dbSpan.SetTag("error.message", err.Error())
            dbSpan.Finish()
            
            span.SetTag("error", true)
            span.SetTag("error.message", err.Error())
            
            http.Error(w, "Internal server error", http.StatusInternalServerError)
            return
        }
        
        dbSpan.SetTag("db.rows_returned", len(products))
        dbSpan.Finish()
        
        // キャッシュに保存
        if len(products) > 0 {
            cacheData, _ := json.Marshal(products)
            s.redis.Set(ctx, cacheKey, cacheData, 5*time.Minute)
        }
    }
    
    // 各製品の詳細情報を非同期で取得
    for i := range products {
        enrichSpan, enrichCtx := tracer.StartSpanFromContext(ctx, "product.enrich")
        
        // 在庫情報の取得
        stockSpan, _ := tracer.StartSpanFromContext(enrichCtx, "inventory.get_stock")
        stockLevel, err := s.getStockLevel(enrichCtx, products[i].ID)
        if err == nil {
            products[i].StockLevel = stockLevel
            stockSpan.SetTag("stock.level", stockLevel)
        } else {
            stockSpan.SetTag("error", true)
            stockSpan.SetTag("error.message", err.Error())
        }
        stockSpan.Finish()
        
        enrichSpan.Finish()
    }
    
    span.SetTag("products.count", len(products))
    span.SetTag("response.status", "success")
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]interface{}{
        "products": products,
        "pagination": map[string]int{
            "page":  page,
            "limit": limit,
            "total": len(products),
        },
    })
}

func (s *Server) getProduct(w http.ResponseWriter, r *http.Request) {
    span, ctx := tracer.StartSpanFromContext(r.Context(), "product.get")
    defer span.Finish()
    
    vars := mux.Vars(r)
    productID, err := strconv.Atoi(vars["id"])
    if err != nil {
        span.SetTag("error", true)
        span.SetTag("error.message", "Invalid product ID")
        
        http.Error(w, "Invalid product ID", http.StatusBadRequest)
        return
    }
    
    span.SetTag("product.id", productID)
    
    // キャッシュから取得を試行
    cacheSpan, ctx := tracer.StartSpanFromContext(ctx, "cache.get")
    cacheKey := fmt.Sprintf("product:%d", productID)
    
    var product Product
    cachedData, err := s.redis.Get(ctx, cacheKey).Result()
    if err == nil {
        cacheSpan.SetTag("cache.hit", true)
        json.Unmarshal([]byte(cachedData), &product)
        span.SetTag("data.source", "cache")
    } else {
        cacheSpan.SetTag("cache.hit", false)
    }
    cacheSpan.Finish()
    
    // データベースから取得
    if product.ID == 0 {
        dbSpan, ctx := tracer.StartSpanFromContext(ctx, "database.query")
        
        query := `
            SELECT id, name, price, category, description, created_at
            FROM products
            WHERE id = $1
        `
        
        dbSpan.SetTag("db.statement", query)
        dbSpan.SetTag("db.operation", "SELECT")
        dbSpan.SetTag("db.table", "products")
        
        err := s.db.GetContext(ctx, &product, query, productID)
        if err != nil {
            if err == sql.ErrNoRows {
                dbSpan.SetTag("db.rows_returned", 0)
                span.SetTag("error.type", "NotFound")
                
                http.Error(w, "Product not found", http.StatusNotFound)
            } else {
                dbSpan.SetTag("error", true)
                dbSpan.SetTag("error.message", err.Error())
                span.SetTag("error", true)
                span.SetTag("error.message", err.Error())
                
                http.Error(w, "Internal server error", http.StatusInternalServerError)
            }
            
            dbSpan.Finish()
            return
        }
        
        dbSpan.SetTag("db.rows_returned", 1)
        dbSpan.Finish()
        
        span.SetTag("data.source", "database")
        
        // キャッシュに保存
        productData, _ := json.Marshal(product)
        s.redis.Set(ctx, cacheKey, productData, 10*time.Minute)
    }
    
    // 関連情報の取得
    relationSpan, relCtx := tracer.StartSpanFromContext(ctx, "product.get_relations")
    
    // 在庫レベル
    stockSpan, _ := tracer.StartSpanFromContext(relCtx, "inventory.get_stock")
    stockLevel, _ := s.getStockLevel(relCtx, product.ID)
    product.StockLevel = stockLevel
    stockSpan.Finish()
    
    // レビュー数
    reviewSpan, _ := tracer.StartSpanFromContext(relCtx, "reviews.get_count")
    reviewCount, averageRating := s.getReviewStats(relCtx, product.ID)
    product.ReviewCount = reviewCount
    product.AverageRating = averageRating
    reviewSpan.SetTag("reviews.count", reviewCount)
    reviewSpan.SetTag("reviews.average_rating", averageRating)
    reviewSpan.Finish()
    
    relationSpan.Finish()
    
    span.SetTag("product.name", product.Name)
    span.SetTag("product.category", product.Category)
    span.SetTag("product.price", product.Price)
    span.SetTag("response.status", "success")
    
    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(product)
}

func (s *Server) getStockLevel(ctx context.Context, productID int) (int, error) {
    span, ctx := tracer.StartSpanFromContext(ctx, "inventory.stock_level")
    defer span.Finish()
    
    span.SetTag("product.id", productID)
    
    var stockLevel int
    query := "SELECT stock_level FROM inventory WHERE product_id = $1"
    
    err := s.db.GetContext(ctx, &stockLevel, query, productID)
    if err != nil {
        span.SetTag("error", true)
        span.SetTag("error.message", err.Error())
        return 0, err
    }
    
    span.SetTag("stock.level", stockLevel)
    return stockLevel, nil
}

func (s *Server) getReviewStats(ctx context.Context, productID int) (int, float64) {
    span, ctx := tracer.StartSpanFromContext(ctx, "reviews.get_stats")
    defer span.Finish()
    
    span.SetTag("product.id", productID)
    
    var count int
    var avgRating float64
    
    query := `
        SELECT COUNT(*) as count, COALESCE(AVG(rating), 0) as avg_rating
        FROM reviews
        WHERE product_id = $1 AND approved = true
    `
    
    row := s.db.QueryRowContext(ctx, query, productID)
    row.Scan(&count, &avgRating)
    
    span.SetTag("reviews.count", count)
    span.SetTag("reviews.avg_rating", avgRating)
    
    return count, avgRating
}