InfluxDB 개념+ JavaClient 구축

환경

- docker

- java

- gradle

개요

사이드 프로젝트에 대한 설계를 진행하였습니다.

저장해야할 데이터들이 시계열 데이터라는 것에 집중하여 이를 저장하기 쉽고 그나마 대중적인 InfluxDB를 사용하자고 얘기가 나오게되었고, RDB에서의 스키마 설계, Client 연동까지 내가 담당하게되었습니다.

데이터베이스를 설계하기 전에, InfluxDB에서 사용하는 언어들에 대한 개념과 그를 토대로 간단한 데이터들을 생성하고 확인하는 테스트 코드를 작성해보았습니다.

내용자체는 (https://cocoder16.tistory.com/83) 이 블로그에 잘 나와있습니다.

그 중 핵심은 RDB와 비교한 그림인데,

RDB	influxDB
database	bucket
table	measurement
column	key
PK or indexed column	tag key
unindexed column	field key
record	point

이런식으로 나타낼 수 있다.

아직까지는 뭔 말인지 모르기에 실제 데이터와 함께 확인해보면 될 것 같습니다.

Influx 1.x vs 2.x

InflxuDB에는 1.x버전과 2.x버전으로 나뉩니다.

GPT말로는 이런 차이가 있다고 하니까 운영하면서 참고하면 될 것 같습니다. 개인적으로 많은 기능을 제공하는 2.x로 프로젝트를 진행할 예정입니다.

java-client에는 2가지 라이브러리가 있습니다

- org.influxdb:influxdb-java : 1.x 버전 특화

- com.influxdb:influxdb-client-java: 2.x 버전 특화

따라서 com.influxdb:influxdb-client-java 라이브러리를 가지고 프로젝트를 구성합니다.

InfluxDB install

InfluxDB는 Docker로 간편하게 설치합니다.

docker run -d --name influxdb -p 8086:8086 influxdb:latest

influxDB는 1.x와 달리 2.x에는 influx setup 명령어를 통해 초기 설정을 해줘야합니다.

도커 컨테이너 내부에 들어가서 해당 명령어를 실행한 뒤, username/ password/ organization/ bucket에 대한 내용을 입력합니다.

이렇게 진해하면 계정이 생성되고, 데이터를 생성할 수 있고 java-client를 연결할 준비는 모두 끝났습니다.

Java Client

라이브러리 추가

dependencies {
    testImplementation platform('org.junit:junit-bom:5.10.0')

    // https://mvnrepository.com/artifact/com.influxdb/influxdb-client-java
    implementation 'com.influxdb:influxdb-client-java:7.2.0'

    testImplementation 'org.junit.jupiter:junit-jupiter'
    compileOnly 'org.projectlombok:lombok:1.18.30'
    annotationProcessor 'org.projectlombok:lombok:1.18.30'

    testCompileOnly 'org.projectlombok:lombok:1.18.30'
    testAnnotationProcessor 'org.projectlombok:lombok:1.18.30'

    // SLF4J API
    implementation 'org.slf4j:slf4j-api:2.0.7'

    // SLF4J 구현체 (Logback)
    implementation 'ch.qos.logback:logback-classic:1.4.8'
}

lombok을 사용하기 위해서 관련 의존성을 추가하고 현재 influxdb-java-client 최신 버전은 7.2.0이라 이 버전을 추가해줍니다.

테스트 코드 작성

@Slf4j
public class InfluxDBConnectionLiveTest {

    // InfluxDB 2.x 접속 정보
    private static final String URL = "http://localhost:8086"; // InfluxDB 서버 주소
    private static final String TOKEN = "UEoxmc--ADNRMXy2x_vj8YDX86DrhU7haqn5lIle8t7zdjK-vuus8SFzuhZTvweSOjflW4bx9HlUBmAQJ41E8A=="; // 생성한 API 토큰
    private static final String ORG = "minseok"; // 생성한 조직 이름
    private static final String BUCKET = "bucket"; // 데이터가 저장될 버킷 이름

    private static InfluxDBClient client;

    @BeforeAll
    public static void setup() {
        // InfluxDB 클라이언트 설정

        client = InfluxDBClientFactory.create(URL, TOKEN.toCharArray(), ORG, BUCKET);
    }

    @Test
    void testWriteData() {
        // WriteApiBlocking을 사용하여 데이터를 InfluxDB에 쓴다.
        WriteApiBlocking writeApi = client.getWriteApiBlocking();
        Point point = Point.measurement("cpu_usage")
                .addTag("host", "server1")
                .addField("value", 75.5)
                .time(Instant.now(), WritePrecision.NS);
        writeApi.writePoint(point);
        System.out.println("✅ 데이터 쓰기 완료!");

        // 데이터가 정상적으로 작성되었는지 확인하는 간단한 테스트
        assertTrue(true, "데이터가 성공적으로 작성되었습니다.");
    }

    @Test
    void testQueryData() {
        // Flux 쿼리 예시
        String fluxQuery = "from(bucket:\"" + BUCKET + "\") |> range(start: -1h)";

        QueryApi queryApi = client.getQueryApi();
        queryApi.query(fluxQuery, (cancellable, record) -> {
            System.out.println("📌 Measurement: " + record.getMeasurement() +
                    ", Field: " + record.getField() +
                    ", Value: " + record.getValue());
        });

        // 데이터가 조회되는지 확인하는 간단한 테스트
        assertTrue(true, "쿼리가 정상적으로 실행되었습니다.");
    }

    @Test
    void testPing() {
        // InfluxDB /health 엔드포인트에 핑을 보내서 연결 상태를 확인
        Boolean ping = client.ping();

        // InfluxDB 서버가 정상적으로 작동 중인지 확인
        assertTrue(ping, "InfluxDB 서버가 정상적으로 작동 중입니다.");
    }

    @Test
    void testDeleteData() {
        // 데이터 삭제 예시 (특정 시간 범위로 삭제)
        String fluxDeleteQuery = "from(bucket:\"" + BUCKET + "\") |> range(start: -1h) |> drop()";

        QueryApi queryApi = client.getQueryApi();
        queryApi.query(fluxDeleteQuery, (cancellable, record) -> {
            // 데이터가 삭제되었는지 확인
            System.out.println("📌 Deleted Record: " + record);
        });

        // 데이터 삭제가 성공적으로 이루어졌는지 확인
        assertTrue(true, "데이터가 정상적으로 삭제되었습니다.");
    }
}

테스트 코드는 간단합니다.

- 커넥션 여부 확인

- 데이터 쓰기 확인

- 데이터 조회 확인

- 데이터 삭제 확인

이런순으로 진행되고 있으며 이렇게 추가된 데이터는 InfluxDB 대시보드에서 확인할 수 있습니다.

위 코드에 상수로 적혀있는 TOKEN은 docker container내부에서 influx auth list 명령어로 확인이 가능하며 그 값을 적어 넣어주시면 됩니다.

테스트 코드를 통해 생성한 값들은 http://localhost:8086에서 확인이 가능합니다.

본인이 생성한 계정을 통해 접속하면

2번째 탭의 Bucket 메뉴를 통해서 본인이 생성한 데이터들을 여러가지 형태로 볼 수 있습니다.

결론

이렇듯 Java와 InfluxDB를 연결하는 방법과 CRD 테스트코드를 작성하여 대시보드에서 확인하는 방법을 확인하였습니다.

좀 더 필요한 학습은 Influx에서 제공하는 여러 개념(Bucket, Organization, field, 등..)에 대한 학습과 여러 인터페이스에 대해서 학습이 필요할 것 같습니다.

ms727 님의 블로그