Time Series Database:収集から分析へ：TSDB for InfluxDB® でデータの価値を創出

Telegraf を使用したメトリクスの収集

Telegraf は Go 言語で記述されたエージェントであり、システムおよびサービスのメトリクスを収集し、HTTP 経由で TSDB for InfluxDB® に書き込みます。Telegraf の構成ファイルには、少なくとも 1 つの 入力プラグイン（メトリクスの取得元）と 1 つの 出力プラグイン（メトリクスの送信先）が必要です。

ステップ 1：Telegraf のインストール

お使いのオペレーティングシステムに応じた Telegraf インストールガイド を参照してください。

ステップ 2：出力プラグインの構成

Telegraf 構成ファイル内で、TSDB for InfluxDB® への書き込みを行うよう [[outputs.influxdb]] セクションを設定します。

[[outputs.influxdb]]
  urls = ["https://xxx.influxdata.rds.aliyuncs.com:3242"]
  database = "test"
  retention_policy = ""   # デフォルトの保持ポリシーを使用する場合は空欄のままにします
  username = "test"
  password = "test"

xxx.influxdata.rds.aliyuncs.com は、インスタンス詳細ページから取得した VPC エンドポイントまたはパブリックエンドポイントに置き換えてください。username および password には、該当データベースに対して書き込み権限を持つアカウント情報を設定してください。

ステップ 3：Telegraf の起動

お使いのオペレーティングシステムに対応するコマンドを実行します。

• Ubuntu、Debian、RedHat、CentOS の場合：

  sudo service telegraf start

• Ubuntu 15.04+、Debian 8+、CentOS 7+、RHEL 7+ の場合：

  sudo systemctl start telegraf

Go SDK を使用したデータの書き込み

GitHub から Go SDK をダウンロードしてください。Go のインストール手順については、Go のドキュメントをご参照ください。

インスタンスのエンドポイントおよび認証情報を使用して HTTPS 接続を行い、バッチ単位でデータを書き込みます。スループットを最大化するために、データサイズに応じて 100～1,000 ポイント／バッチで書き込んでください。

package main

import (
    "fmt"
    "log"
    "math/rand"
    "net/url"
    "time"

    client "github.com/influxdata/influxdb1-client"
)

func main() {
    // HTTPS 経由でインスタンスエンドポイントに接続
    host, err := url.Parse(fmt.Sprintf("https://%s:%d", "xxx.influxdata.rds.aliyuncs.com", 3242))
    if err != nil {
        log.Fatal(err)
    }
    config := client.Config{
        URL:      *host,
        Username: "test",
        Password: "test",
    }
    con, err := client.NewClient(config)
    if err != nil {
        log.Fatal(err)
    }

    _, _, err = con.Ping()
    if err != nil {
        log.Fatal(err)
    }

    // 1,000 ポイントのバッチを構築
    var (
        shapes     = []string{"circle", "rectangle", "square", "triangle"}
        colors     = []string{"red", "blue", "green"}
        sampleSize = 1000
        pts        = make([]client.Point, sampleSize)
    )

    rand.Seed(42)
    for i := 0; i < sampleSize; i++ {
        pts[i] = client.Point{
            Measurement: "shapes",
            Tags: map[string]string{
                "color": colors[rand.Intn(len(colors))],
                "shape": shapes[rand.Intn(len(shapes))],
            },
            Fields: map[string]interface{}{
                "value": rand.Intn(sampleSize),
            },
            Time: time.Now(),
        }
    }

    // 対象データベースへバッチを書き込み
    bps := client.BatchPoints{
        Points:          pts,
        Database:        "test",
        RetentionPolicy: "autogen",
    }
    _, err = con.Write(bps)
    if err != nil {
        log.Fatal(err)
    }
}

コンパイルおよび実行方法：

go run main.go

Java SDK を使用したデータの書き込み

GitHub から Java SDK をダウンロードしてください。

package main;

import java.util.concurrent.TimeUnit;
import org.influxdb.InfluxDB;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;

public class StartMain {

    public static void main(String[] args) throws Exception {
        StartMain startMain = new StartMain();
        try {
            startMain.run();
        } catch (Exception e) {
            System.out.println(e);
        }
    }

    public void run() throws Exception {
        // HTTPS 経由でインスタンスエンドポイントに接続
        InfluxDB influxDB = InfluxDBFactory.connect(
            "https://xxx.influxdata.rds.aliyuncs.com:3242", "test", "test");

        String dbName = "test";
        influxDB.setDatabase(dbName);
        String rpName = "autogen";
        influxDB.setRetentionPolicy(rpName);
        influxDB.enableBatch(BatchOptions.DEFAULTS);

        // CPU メトリクスの書き込み
        influxDB.write(Point.measurement("cpu")
            .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
            .addField("idle", 90L)
            .addField("user", 9L)
            .addField("system", 1L)
            .build());

        // ディスクメトリクスの書き込み
        influxDB.write(Point.measurement("disk")
            .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS)
            .addField("used", 80L)
            .addField("free", 1L)
            .build());

        influxDB.close();
    }
}

書き込みエラーを処理するには、enableBatch() にカスタム例外ハンドラーを渡します。

influxDB.enableBatch(BatchOptions.DEFAULTS.exceptionHandler(
    (failedPoints, throwable) -> { /* ここにカスタムエラー処理を記述 */ })
);

Influx CLI を使用したデータの書き込み

-ssl フラグとともにホスト、ポート、ユーザー名、パスワードを指定して接続します。

influx -ssl -host xxx.influxdata.rds.aliyuncs.com -port 3242 -username admin -password admin

TSDB for InfluxDB® へクライアントからデータを書き込むには、「ラインプロトコルチュートリアル」をご参照ください。

ラインプロトコルでデータポイントを書き込みます。

> INSERT cpu,host=serverA,region=us_west value=0.64

HTTP API を使用したデータの書き込み

curl を使用して、ラインプロトコル形式のデータを書き込みエンドポイントに直接 POST します。

curl -i -XPOST -u test:test \
  'https://xxx.influxdata.rds.aliyuncs.com:3242/write?db=test' \
  --data-binary 'cpu_load_short,host=server01,region=us-west value=0.64 1434055562000000000'