> For the complete documentation index, see [llms.txt](https://kopens.gitbook.io/plantpulse-platform/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://kopens.gitbook.io/plantpulse-platform/developer/report.md).

# 리포트 데이터 로직

## 개요 <a href="#overview" id="overview"></a>

이 문서에서는 플랜트펄스 플랫폼의 리포트 데이터 수집 및 저장 로직을 안내합니다. 리포트 시스템은 인메모리 카운터를 통해 실시간으로 데이터를 집계하고, 1분 주기로 Cassandra에 플러시하는 구조로 설계되어 있습니다. 이를 통해 파이프라인 성능에 영향을 주지 않으면서도 정확한 통계 데이터를 확보합니다.

## 데이터 흐름 다이어그램 <a href="#data-flow-diagram" id="data-flow-diagram"></a>

```
┌──────────────┐     ┌──────────────────┐     ┌───────────────────┐
│  Pipeline    │     │  ReportCounter   │     │  Cassandra        │
│  DataFlow    │     │  (In-Memory)     │     │  (tm_report_*)    │
│              │     │                  │     │                   │
│  Prepare  ───┼──▶  │  counter++       │     │                   │
│  Validate ───┼──▶  │  (AtomicLong)    │     │                   │
│  Cache    ───┼──▶  │                  │     │                   │
│  Stream   ───┼──▶  │                  │     │                   │
│  Store    ───┼──▶  │                  │     │                   │
│  DDS      ───┼──▶  │                  │     │                   │
│              │     │                  │     │                   │
└──────────────┘     │  ┌────────────┐  │     │                   │
                     │  │ ReportJob  │──┼──▶  │  INSERT/UPDATE    │
                     │  │ (1분 주기)  │  │     │  (배치 쓰기)       │
                     │  └────────────┘  │     │                   │
                     └──────────────────┘     └───────────────────┘
                                                       │
                                                       ▼
                                              ┌───────────────────┐
                                              │  Dashboard        │
                                              │  (차트/그래프)      │
                                              └───────────────────┘
```

## 카운터 종류 <a href="#counter-types" id="counter-types"></a>

### ReportCounter

`ReportCounter`는 `AtomicLong` 기반의 인메모리 카운터로, 12가지 타입을 관리합니다.

| #  | 카운터 타입            | 키 패턴                       | 설명                      |
| -- | ----------------- | -------------------------- | ----------------------- |
| 1  | `INGESTED`        | `{tag_id}.ingested`        | 수신된 데이터 포인트 수입니다        |
| 2  | `PROCESSED`       | `{tag_id}.processed`       | 처리 완료된 데이터 포인트 수입니다     |
| 3  | `DROPPED`         | `{tag_id}.dropped`         | 검증 실패로 드롭된 데이터 수입니다     |
| 4  | `STORED`          | `{tag_id}.stored`          | Cassandra에 저장된 데이터 수입니다 |
| 5  | `PUBLISHED`       | `{tag_id}.published`       | DDS로 발행된 데이터 수입니다       |
| 6  | `TIMEOUT`         | `{tag_id}.timeout`         | 타임아웃 발생 건수입니다           |
| 7  | `ERROR`           | `{tag_id}.error`           | 에러 발생 건수입니다             |
| 8  | `ALARM_TRIGGERED` | `{tag_id}.alarm_triggered` | 알람 발생 건수입니다             |
| 9  | `ALARM_CLEARED`   | `{tag_id}.alarm_cleared`   | 알람 해제 건수입니다             |
| 10 | `CEP_MATCHED`     | `{tag_id}.cep_matched`     | CEP 규칙 매칭 건수입니다         |
| 11 | `CACHE_HIT`       | `{tag_id}.cache_hit`       | 캐시 히트 건수입니다             |
| 12 | `CACHE_MISS`      | `{tag_id}.cache_miss`      | 캐시 미스 건수입니다             |

모든 카운터는 `AtomicLong`을 사용하여 Lock-free 방식으로 증분되므로, 파이프라인 처리 성능에 미치는 영향이 극히 미미합니다.

## 카운터 증분 시점 <a href="#counter-increment-timing" id="counter-increment-timing"></a>

각 카운터가 증분되는 위치와 트리거 조건입니다.

| 카운터 타입            | 호출 위치                           | 트리거                          |
| ----------------- | ------------------------------- | ---------------------------- |
| `INGESTED`        | `PipelineEngine.enqueue()`      | 데이터가 파이프라인에 투입될 때 증분됩니다      |
| `PROCESSED`       | `DataFlowExecutor.execute()`    | DataFlow 6단계를 모두 통과한 후 증분됩니다 |
| `DROPPED`         | `ValidateStage.validate()`      | 검증 실패 시 증분됩니다                |
| `STORED`          | `StoreStage.store()`            | Cassandra 쓰기 성공 시 증분됩니다      |
| `PUBLISHED`       | `DDSStage.publish()`            | DDS 발행 성공 시 증분됩니다            |
| `TIMEOUT`         | `TimeoutBackupManager.backup()` | 처리 타임아웃 발생 시 증분됩니다           |
| `ERROR`           | `DataFlowExecutor.onError()`    | 스테이지 실행 중 예외 발생 시 증분됩니다      |
| `ALARM_TRIGGERED` | `AlarmEvaluator.evaluate()`     | 알람 조건이 충족되어 알람이 발생할 때 증분됩니다  |
| `ALARM_CLEARED`   | `AlarmEvaluator.clear()`        | 알람 조건이 해제될 때 증분됩니다           |
| `CEP_MATCHED`     | `CEPEngine.onMatch()`           | CEP 규칙이 매칭될 때 증분됩니다          |
| `CACHE_HIT`       | `CacheStage.lookup()`           | 캐시 조회 시 히트할 때 증분됩니다          |
| `CACHE_MISS`      | `CacheStage.lookup()`           | 캐시 조회 시 미스할 때 증분됩니다          |

## ReportJob 플러시 로직 <a href="#report-job-flush" id="report-job-flush"></a>

`ReportJob`은 Quartz 스케줄러에 의해 1분 주기로 실행되며, 인메모리 카운터를 Cassandra에 플러시합니다. 플러시는 5단계로 진행됩니다.

### 5단계 플러시 프로세스

```
ReportJob.execute()
    │
    ├── Step 1: 읽기 + 리셋 (getAndReset)
    │   └── 모든 카운터의 현재값을 읽고 0으로 리셋합니다
    │
    ├── Step 2: 태그별 집계 → tm_report_tag
    │   └── 태그 단위 카운터를 Cassandra에 저장합니다
    │
    ├── Step 3: OPC별 합산 → tm_report_opc
    │   └── 동일 OPC에 속한 태그 카운터를 합산하여 저장합니다
    │
    ├── Step 4: 에셋별 합산 → tm_report_asset
    │   └── 동일 에셋에 속한 태그 카운터를 합산하여 저장합니다
    │
    └── Step 5: 계층별 합산 → tm_report_line/area/site/summary
        └── Line → Area → Site → Summary 순으로 상위 계층에 합산합니다
```

### Step 1: 읽기 + 리셋 (getAndReset)

`AtomicLong.getAndSet(0)`을 사용하여 현재 카운터 값을 원자적으로 읽고 0으로 리셋합니다. 이 방식은 Lock-free이며, 읽기와 리셋 사이에 유입되는 데이터의 카운트 누락이 발생하지 않습니다.

### Step 2\~4: 태그 → OPC → 에셋 집계

```
태그 카운터 (개별)
    │
    ├──▶ tm_report_tag   (태그별 저장)
    │
    ├── 동일 OPC 그룹핑 ──▶ tm_report_opc   (OPC별 합산)
    │
    └── 동일 에셋 그룹핑 ──▶ tm_report_asset (에셋별 합산)
```

### Step 5: 계층별 합산

에셋 카운터를 기반으로 상위 계층까지 순차적으로 합산합니다.

```
tm_report_asset (에셋)
    │
    ├── Line 그룹핑 ──▶ tm_report_line (라인별 합산)
    │
    ├── Area 그룹핑 ──▶ tm_report_area (구역별 합산)
    │
    ├── Site 그룹핑 ──▶ tm_report_site (사이트별 합산)
    │
    └── 전체 합산 ──▶ tm_report_summary (전체 요약)
```

## Cassandra 테이블 구조 <a href="#cassandra-tables" id="cassandra-tables"></a>

리포트 데이터는 계층별로 분리된 Cassandra 테이블에 저장됩니다.

### tm\_report\_tag

태그 단위의 가장 세밀한 리포트 데이터입니다.

| 컬럼                | 타입          | 설명                  |
| ----------------- | ----------- | ------------------- |
| `tag_id`          | `text`      | 태그 ID (파티션 키)입니다    |
| `report_time`     | `timestamp` | 리포트 시간 (클러스터링 키)입니다 |
| `ingested`        | `bigint`    | 수신 건수입니다            |
| `processed`       | `bigint`    | 처리 건수입니다            |
| `dropped`         | `bigint`    | 드롭 건수입니다            |
| `stored`          | `bigint`    | 저장 건수입니다            |
| `published`       | `bigint`    | 발행 건수입니다            |
| `timeout`         | `bigint`    | 타임아웃 건수입니다          |
| `error`           | `bigint`    | 에러 건수입니다            |
| `alarm_triggered` | `bigint`    | 알람 발생 건수입니다         |
| `alarm_cleared`   | `bigint`    | 알람 해제 건수입니다         |
| `cep_matched`     | `bigint`    | CEP 매칭 건수입니다        |

### tm\_report\_opc

OPC 연결 단위의 합산 리포트입니다.

| 컬럼                          | 타입          | 설명                  |
| --------------------------- | ----------- | ------------------- |
| `opc_id`                    | `text`      | OPC ID (파티션 키)입니다   |
| `report_time`               | `timestamp` | 리포트 시간 (클러스터링 키)입니다 |
| `ingested` \~ `cep_matched` | `bigint`    | 소속 태그의 합산 값입니다      |
| `tag_count`                 | `int`       | 소속 태그 수입니다          |

### tm\_report\_asset

에셋(설비) 단위의 합산 리포트입니다.

| 컬럼                          | 타입          | 설명                  |
| --------------------------- | ----------- | ------------------- |
| `asset_id`                  | `text`      | 에셋 ID (파티션 키)입니다    |
| `report_time`               | `timestamp` | 리포트 시간 (클러스터링 키)입니다 |
| `ingested` \~ `cep_matched` | `bigint`    | 소속 태그의 합산 값입니다      |
| `tag_count`                 | `int`       | 소속 태그 수입니다          |

### tm\_report\_line / tm\_report\_area / tm\_report\_site

라인, 구역, 사이트 단위의 합산 리포트입니다. 각 테이블은 해당 계층의 ID를 파티션 키로 사용하며, 하위 계층 데이터를 합산한 값을 저장합니다.

### tm\_report\_summary

전체 플랫폼의 요약 리포트입니다.

| 컬럼                          | 타입          | 설명                         |
| --------------------------- | ----------- | -------------------------- |
| `summary_key`               | `text`      | 고정 키 `'GLOBAL'` (파티션 키)입니다 |
| `report_time`               | `timestamp` | 리포트 시간 (클러스터링 키)입니다        |
| `ingested` \~ `cep_matched` | `bigint`    | 전체 합산 값입니다                 |
| `total_tags`                | `int`       | 전체 활성 태그 수입니다              |
| `total_opcs`                | `int`       | 전체 활성 OPC 수입니다             |
| `total_assets`              | `int`       | 전체 활성 에셋 수입니다              |

## 포인트 카운트 데이터 흐름 <a href="#point-count-flow" id="point-count-flow"></a>

포인트 카운트는 리포트 카운터와는 별도로, 실시간 데이터 수신 건수를 추적하는 경량 메커니즘입니다.

```
데이터 포인트 수신
    │
    ▼
PointCountMap (ConcurrentHashMap)
    │  key: tag_id
    │  value: AtomicLong (수신 건수)
    │
    ▼ (1분 주기 - PointCountFlushJob)
    │
    ├──▶ Redis 갱신 (실시간 현재값)
    │
    └──▶ Cassandra 저장 (이력)
            │
            ▼
      tm_point_count 테이블
```

| 항목     | 설명                                                          |
| ------ | ----------------------------------------------------------- |
| 수집 방식  | `ConcurrentHashMap<String, AtomicLong>` 기반 Lock-free 카운팅입니다 |
| 플러시 주기 | 1분 (PointCountFlushJob)입니다                                  |
| 저장소    | Redis (실시간) + Cassandra (이력)입니다                             |
| 용도     | 태그별 데이터 수신 빈도 모니터링, 비활성 태그 감지에 활용됩니다                        |

## 대시보드 차트 데이터 소스 <a href="#dashboard-data-source" id="dashboard-data-source"></a>

관리 콘솔 대시보드의 차트는 아래 Cassandra 테이블을 데이터 소스로 사용합니다.

| 차트         | 데이터 소스 테이블          | 조회 범위            |
| ---------- | ------------------- | ---------------- |
| 실시간 수신 건수  | `tm_report_summary` | 최근 1시간 (1분 단위)   |
| 사이트별 처리량   | `tm_report_site`    | 최근 24시간 (1시간 단위) |
| OPC별 수신 현황 | `tm_report_opc`     | 최근 1시간 (1분 단위)   |
| 태그별 상세 통계  | `tm_report_tag`     | 최근 1시간 (1분 단위)   |
| 에셋별 알람 통계  | `tm_report_asset`   | 최근 24시간 (1시간 단위) |
| 에러율 추이     | `tm_report_summary` | 최근 7일 (1일 단위)    |
| 포인트 수신 빈도  | `tm_point_count`    | 최근 1시간 (1분 단위)   |

대시보드는 REST API를 통해 위 테이블의 데이터를 조회하며, 프론트엔드에서 시계열 차트로 시각화합니다.