InfluxDB

// InfluxDB Node.js 클라이언트 예제
const { InfluxDB, Point } = require('@influxdata/influxdb-client');

// 연결 설정
const url = 'http://localhost:8086';
const token = 'your-token-here';
const org = 'your-org';
const bucket = 'metrics';

const client = new InfluxDB({ url, token });

// ============================================
// 데이터 쓰기
// ============================================
const writeApi = client.getWriteApi(org, bucket);

// 기본 태그 설정 (모든 포인트에 적용)
writeApi.useDefaultTags({ app: 'my-app', env: 'production' });

// Point 객체로 데이터 생성
const point = new Point('api_metrics')
    .tag('endpoint', '/api/users')
    .tag('method', 'GET')
    .floatField('response_time', 125.5)
    .intField('status_code', 200)
    .intField('request_size', 1024);

writeApi.writePoint(point);

// 여러 포인트 쓰기
for (let i = 0; i < 10; i++) {
    const p = new Point('request_log')
        .tag('service', 'user-service')
        .floatField('latency', Math.random() * 100)
        .intField('requests', Math.floor(Math.random() * 1000));
    writeApi.writePoint(p);
}

// 버퍼 플러시 및 완료 대기
writeApi.close()
    .then(() => console.log('데이터 기록 완료'))
    .catch(err => console.error('쓰기 오류:', err));


// ============================================
// 데이터 조회
// ============================================
const queryApi = client.getQueryApi(org);

const fluxQuery = `
from(bucket: "${bucket}")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "api_metrics")
    |> filter(fn: (r) => r._field == "response_time")
`;

console.log('\n쿼리 결과:');
queryApi.queryRows(fluxQuery, {
    next: (row, tableMeta) => {
        const record = tableMeta.toObject(row);
        console.log(`  ${record._time}: ${record._value}ms (${record.endpoint})`);
    },
    error: (err) => console.error('쿼리 오류:', err),
    complete: () => console.log('쿼리 완료'),
});


// ============================================
// Promise 기반 조회
// ============================================
async function queryMetrics() {
    const query = `
    from(bucket: "${bucket}")
        |> range(start: -24h)
        |> filter(fn: (r) => r._measurement == "api_metrics")
        |> aggregateWindow(every: 1h, fn: mean)
        |> sort(columns: ["_time"], desc: true)
        |> limit(n: 10)
    `;

    const results = [];

    for await (const { values, tableMeta } of queryApi.iterateRows(query)) {
        const record = tableMeta.toObject(values);
        results.push({
            time: record._time,
            field: record._field,
            value: record._value,
            endpoint: record.endpoint
        });
    }

    return results;
}

queryMetrics()
    .then(data => {
        console.log('\n시간별 평균 응답시간:');
        data.forEach(r => console.log(`  ${r.time}: ${r.value?.toFixed(2)}ms`));
    });


// ============================================
// Health Check
// ============================================
const { HealthAPI } = require('@influxdata/influxdb-client-apis');

const healthApi = new HealthAPI(client);
healthApi.getHealth()
    .then(health => console.log('\nInfluxDB 상태:', health.status))
    .catch(err => console.error('Health check 실패:', err));

// ============================================
// Flux 쿼리 언어 기본
// ============================================

// 기본 조회 - 최근 1시간 데이터
from(bucket: "metrics")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "server_metrics")

// 특정 필드와 태그 필터링
from(bucket: "metrics")
    |> range(start: -24h)
    |> filter(fn: (r) => r._measurement == "cpu")
    |> filter(fn: (r) => r._field == "usage_percent")
    |> filter(fn: (r) => r.host == "server-01" or r.host == "server-02")


// ============================================
// 집계 함수
// ============================================

// 윈도우 집계 - 5분 평균
from(bucket: "metrics")
    |> range(start: -6h)
    |> filter(fn: (r) => r._measurement == "cpu")
    |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)

// 다양한 집계 함수
from(bucket: "metrics")
    |> range(start: -1h)
    |> filter(fn: (r) => r._field == "temperature")
    |> aggregateWindow(every: 10m, fn: max)  // max, min, sum, count, last

// 그룹별 집계
from(bucket: "metrics")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "api_requests")
    |> group(columns: ["endpoint"])
    |> count()


// ============================================
// 다운샘플링 & 데이터 변환
// ============================================

// 다운샘플링 - 1분 데이터를 1시간으로
from(bucket: "raw_metrics")
    |> range(start: -7d)
    |> aggregateWindow(every: 1h, fn: mean)
    |> to(bucket: "downsampled_metrics")

// 이동 평균 (Moving Average)
from(bucket: "metrics")
    |> range(start: -2h)
    |> filter(fn: (r) => r._field == "cpu_usage")
    |> movingAverage(n: 5)

// 변화율 계산 (derivative)
from(bucket: "metrics")
    |> range(start: -1h)
    |> filter(fn: (r) => r._field == "requests_total")
    |> derivative(unit: 1m, nonNegative: true)


// ============================================
// 조인 & 피벗
// ============================================

// 두 measurement 조인
cpu = from(bucket: "metrics")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "cpu")

memory = from(bucket: "metrics")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "memory")

join(tables: {cpu: cpu, memory: memory}, on: ["_time", "host"])

// 피벗 - 필드를 컬럼으로 변환
from(bucket: "metrics")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "server_metrics")
    |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")


// ============================================
// 알림 조건
// ============================================

// CPU 80% 이상 감지
from(bucket: "metrics")
    |> range(start: -5m)
    |> filter(fn: (r) => r._field == "cpu_usage")
    |> filter(fn: (r) => r._value > 80.0)
    |> count()
    |> map(fn: (r) => ({r with alert: r._value > 0}))