Prometheus Metrics Sink Connector 構成プロパティ¶

このコネクターを使用するには、connector.class 構成プロパティでこのコネクタークラスの名前を指定します。

connector.class=io.confluent.connect.prometheus.PrometheusMetricsSinkConnector

コネクター固有の構成プロパティについて、以降で説明します。

コネクター¶

prometheus.listener.url

メトリクスをスクレイピングするために Prometheus 用 HTTP サーバーによって公開される URL。

型: string
デフォルト : http://localhost:8889/metrics
指定可能な値 : 次のいずれかのスキーム http を使用した URI
重要度: 中

behavior.on.error

エラー処理の動作の設定。以下のいずれかに構成する必要があります。

fail

メトリクスの処理中またはメトリックバッファへのメトリクスの挿入中にエラーが発生すると、コネクターによって ConnectException がスローされ、メトリクスの処理が停止されます。

ignore

メトリクスの処理中またはメトリックバッファへのメトリクスの挿入中にエラーが発生すると、次のメトリクスセットの処理が続行されます。

log

メトリクスの処理中またはメトリックバッファへのメトリクスの挿入中にエラーが発生すると、connect-reporter 経由でエラーが記録され、次のレコードセットの処理が続行されます。記録先は ${connector-name}-error トピックです。

型: string
デフォルト: fail
指定可能な値: [log, ignore, fail] のいずれか
重要度: 中

max.retry.time.ms

Connect では、Prometheus によってバッファからメトリクスがスクレイピング中であっても、スクレイピングの終了が待機され、終了するとバッファに別のレコードが追加されます。 max.retry.time.ms はコネクターによってバッファへのレコードの追加が再試行される期間（ミリ秒）で、これを超えるとエラーになります。

型: long
デフォルト: 60000
指定可能な値: [60000,…]
重要度: 低

TLS¶

prometheus.listener.ssl.keystore.location

HTTPS を利用して Prometheus エンドポイント用に使用します。SSL に使用するキーストアファイルの場所。

型: string
デフォルト: ""
重要度: 低

prometheus.listener.ssl.keystore.password

HTTPS を利用して Prometheus エンドポイント用に使用します。キーストアファイルのストアパスワード。

型: password
デフォルト: [hidden]
重要度: 中

prometheus.listener.ssl.key.password

HTTPS を利用して Prometheus エンドポイント用に使用します。キーストアファイルのプライベートキーのパスワード。

型: password
デフォルト: [hidden]
重要度: 中

基本認証¶

prometheus.listener.basic.auth.username

基本アクセス認証用のユーザー名。このユーザー名は、HTTP ユーザーエージェント（Prometheus）によってメトリクスの取得がリクエストされる際、認証に使用されます。

型: string
デフォルト: ""
重要度: 中

prometheus.listener.basic.auth.password

基本アクセス認証用のパスワード。このパスワードは、HTTP ユーザーエージェント（Prometheus）によってメトリクスの取得がリクエストされる際、認証に使用されます。

型: password
デフォルト: [hidden]
重要度: 中

Connect Reporter¶

Reporter の詳細については、「Connect Reporter 」を参照してください。

reporter.result.topic.name

シンクレコードの処理が成功した後のレコードの生成先となるトピックの名前。現在のコネクター名を指定するには、パターン内で ${connector} を使用します。エラーレポートの動作を無効にするには、空白のままにします。

型: string
デフォルト: ${connector}-success
指定可能な値 : ${connector} を置き換える場合は、ASCII 英数字、+、.、_、- の文字から 1 ～ 249 個を使用して有効なトピック名を指定する必要があります。
重要度: 中

reporter.result.topic.replication.factor

このコネクターによって自動的に作成される場合の結果トピックのレプリケーション係数。これは、許容可能なブローカー障害数を指定するものです。これを超えるとデータが失われます。これは、開発環境では 1、本稼働環境では "常に" 3 以上にする必要があります。

型: short
デフォルト: 3
指定可能な値: [1,…]
重要度: 中

reporter.result.topic.partitions

このコネクターによって自動的に作成される場合の結果トピックのパーティション数。このパーティション数で、潜在的なスループットを処理するには、入力パーティション数と同じにする必要があります。

型: int
デフォルト: 1
指定可能な値: [1,…]
重要度: 中

reporter.error.topic.name

レコードのシンク試行が失敗した後にレコードの生成先となるトピックの名前。現在のコネクター名を指定するには、パターン内で ${connector} を使用します。エラーレポートの動作を無効にするには、空白のままにします。

型: string
デフォルト: ${connector}-error
指定可能な値 : ${connector} を置き換える場合は、ASCII 英数字、+、.、_、- の文字から 1 ～ 249 個を使用して有効なトピック名を指定する必要があります。
重要度: 中

reporter.error.topic.replication.factor

このコネクターによって自動的に作成される場合のエラートピックのレプリケーション係数。これは、許容可能なブローカー障害数を指定するものです。これを超えるとデータが失われます。これは、開発環境では 1、本稼働環境では "常に" 3 以上にする必要があります。

型: short
デフォルト: 3
指定可能な値: [1,…]
重要度: 中

reporter.error.topic.partitions

このコネクターによって自動的に作成される場合のエラートピックのパーティション数。このパーティション数で、潜在的なスループットを処理するには、入力パーティション数と同じにする必要があります。

型: int
デフォルト: 1
指定可能な値: [1,…]
重要度: 中

reporter.bootstrap.servers

Kafka クラスターとの初期接続を確立するために使用するホストとポートのペアのリスト。クライアントは、ここで指定されるブートストラップサーバーに関係なくすべてのサーバーを使用します。このリストは、すべてのサーバーセットを検出するのに使用される初期ホストにのみ影響します。このリストは、host1:port1,host2:port2,… という形式にする必要があります。これらのサーバーでは、初期接続ですべてのクラスターメンバーシップを検出するためにのみ使用されます。これは動的に変わる可能性があるので、このリストにすべてのサーバーセットを含める必要はありません。ただし、サーバーの障害に備えて、複数指定しておくこともできます。

型: list
指定可能な値: 空ではないリスト
重要度: 高

フォーマッター¶

reporter.result.topic.key.format

結果レポートキーがシリアル化されるフォーマット。

型: string
デフォルト: json
指定可能な値: [string, json] のいずれか
重要度: 中
依存関係 : reporter.result.topic.key.format.schemas.enable、reporter.result.topic.key.format.schemas.cache.size

reporter.result.topic.value.format

結果レポート値がシリアル化されるフォーマット。

型: string
デフォルト: json
指定可能な値: [string, json] のいずれか
重要度: 中
依存関係 : reporter.result.topic.value.format.schemas.cache.size、reporter.result.topic.value.format.schemas.enable

reporter.error.topic.key.format

エラーレポートキーがシリアル化されるフォーマット。

型: string
デフォルト: json
指定可能な値: [string, json] のいずれか
重要度: 中
依存関係 : reporter.error.topic.key.format.schemas.cache.size、reporter.error.topic.key.format.schemas.enable

reporter.error.topic.value.format

エラーレポート値がシリアル化されるフォーマット。

型: string
デフォルト: json
指定可能な値: [string, json] のいずれか
重要度: 中
依存関係 : reporter.error.topic.value.format.schemas.cache.size、reporter.error.topic.value.format.schemas.enable

JSON フォーマッター¶

reporter.result.topic.key.format.schemas.cache.size

JSON フォーマッターでキャッシュできる最大スキーマ数。

型: int
デフォルト: 128
指定可能な値: [0,…,2048]
重要度: 中

reporter.result.topic.key.format.schemas.enable

シリアル化された値とキーのそれぞれにスキーマを含めます。

型: boolean
デフォルト: false
重要度: 中

reporter.result.topic.value.format.schemas.cache.size

JSON フォーマッターでキャッシュできる最大スキーマ数。

型: int
デフォルト: 128
指定可能な値: [0,…,2048]
重要度: 中

reporter.result.topic.value.format.schemas.enable

シリアル化された値とキーのそれぞれにスキーマを含めます。

型: boolean
デフォルト: false
重要度: 中

reporter.error.topic.key.format.schemas.cache.size

JSON フォーマッターでキャッシュできる最大スキーマ数。

型: int
デフォルト: 128
指定可能な値: [0,…,2048]
重要度: 中

reporter.error.topic.key.format.schemas.enable

シリアル化された値とキーのそれぞれにスキーマを含めます。

型: boolean
デフォルト: false
重要度: 中

reporter.error.topic.value.format.schemas.cache.size

JSON フォーマッターでキャッシュできる最大スキーマ数。

型: int
デフォルト: 128
指定可能な値: [0,…,2048]
重要度: 中

reporter.error.topic.value.format.schemas.enable

シリアル化された値とキーのそれぞれにスキーマを含めます。

型: boolean
デフォルト: false
重要度: 中

Confluent Platform ライセンス¶

confluent.license

Confluent では、各契約者にライセンスキーを発行します。ライセンスキーは、コピーアンドペーストできる短いテキストです。このコネクターは、ライセンスキーがなくても 30 日間試用できます。既にご契約されている場合は、詳細について Confluent サポートにお問い合わせください。

型: string
デフォルト: ""
指定可能な値: Confluent Platform ライセンス
重要度: 高

confluent.topic.bootstrap.servers

ライセンス供与に使用される Kafka クラスターとの初期接続を確立するために使用するホストとポートのペアのリスト。初期接続で、クラスター内のすべてのサーバーが検出されます。このリストは、host1:port1,host2:port2,… という形式にする必要があります。これらのサーバーは、初期接続ですべてのクラスターメンバーシップを検出するためにのみ使用されます。これは動的に変わる可能性があるので、このリストにすべてのサーバーセットを含める必要はありません。サーバーの障害に備えて、複数指定しておくこともできます。

型: list
重要度: 高

confluent.topic

Confluent Platform の構成（ライセンス情報など）で使用される Kafka のトピックの名前。

型: string
デフォルト: _confluent-command
重要度: 低

confluent.topic.replication.factor

Confluent Platform の構成（ライセンス情報など）で使用される Kafka のトピックのレプリケーション係数。これは、トピックがまだ存在しない場合にのみ使用されます。デフォルトの 3 は、本稼働環境での使用に適しています。ブローカー数が 3 未満の開発環境で使用する場合は、そのブローカー数をこのプロパティに設定する必要があります（通常は 1 ）。

型: int
デフォルト: 3
重要度: 低

Confluent ライセンスのプロパティ¶

ちなみに

コネクターの構成にライセンス関連のプロパティを含めることができますが、Confluent Platform バージョン 6.0 以降では、各コネクターの構成ではなく、Connect ワーカーの構成にライセンス関連のプロパティを設定できるようになりました。

注釈

このコネクターはプロプライエタリであり、ライセンスが必要です。ライセンス情報は、_confluent-command トピックに格納されます。ブローカーが SSL 接続を必要とする場合は、以下で説明するセキュリティ関連の confluent.topic.* プロパティを含める必要があります。

confluent.license

Confluent では、各契約者にエンタープライズライセンスキーを発行します。ライセンスキーはテキストであるため、confluent.license の値としてコピーアンドペーストできます。試用ライセンスでは、コネクターを 30 日間使用できます。開発者ライセンスでは、ブローカーが 1 つの開発環境で、コネクターを無期限に使用できます。

既にご契約されている場合は、詳細について Confluent サポートにお問い合わせください。

型: string
デフォルト: ""
指定可能な値: Confluent Platform ライセンス
重要度: 高

confluent.topic.ssl.truststore.location

トラストストアファイルの場所。

型: string
デフォルト: null
重要度: 高

confluent.topic.ssl.truststore.password

トラストストアファイルのパスワード。パスワードが設定されていなくてもトラストストアにアクセスできますが、整合性チェックが無効になります。

型: password
デフォルト: null
重要度: 高

confluent.topic.ssl.keystore.location

キーストアファイルの場所。クライアントでは省略可能です。クライアントの相互認証に使用できます。

型: string
デフォルト: null
重要度: 高

confluent.topic.ssl.keystore.password

キーストアファイルのストアパスワード。クライアントでは省略可能です。ssl.keystore.location を構成した場合にのみ必要となります。

型: password
デフォルト: null
重要度: 高

confluent.topic.ssl.key.password

キーストアファイル内のプライベートキーのパスワード。クライアントでは省略可能です。

型: password
デフォルト: null
重要度: 高

confluent.topic.security.protocol

ブローカーとの通信に使用されるプロトコル。指定可能な値は、PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL です。

型: string
デフォルト: "PLAINTEXT"
重要度: 中

ライセンストピックの構成¶

Confluent エンタープライズライセンスは、_confluent-command トピックに格納されます。このトピックでは、デフォルトで作成され、confluent.license プロパティで指定されたライセンスキーに対応するライセンスが含まれます。

注釈

パブリックキーは Kafka のトピックには保管されません。

以下では、デフォルトの _confluent-command トピックがさまざまなシナリオでどのように生成されるかを説明します。

confluent.license プロパティを追加していない場合、またはこのプロパティを（たとえば、confluent.license= ）空にした場合は、_confluent command トピックに、30 日間の試用ライセンスが自動的に生成されます。
有効なライセンスキーを（たとえば、confluent.license=<valid-license-key> ）追加すると、有効なライセンスが _confluent-command トピックに追加されます。

以下は、開発およびテストのための最小限のプロパティの例です。

_confluent-command トピックの名前は confluent.topic プロパティを使用して変更できます（たとえば、環境に厳格な命名規則がある場合など）。以下の例は、この変更および構成された Kafka ブートストラップサーバーを示しています。

confluent.topic=foo_confluent-command
confluent.topic.bootstrap.servers=localhost:9092

上記の例は、開発およびテスト用に使用できるブートストラップサーバーの必要最小限のプロパティを示しています。本稼働環境の場合は、プレフィックスとして confluent.topic. を付けて、通常のプロデューサー、コンシューマー、およびトピックの各構成プロパティをコネクターのプロパティに追加します。

ライセンストピックの ACL¶

_confluent-command トピックには、confluent.license プロパティで指定されたライセンスキーに対応するライセンスが含まれます。このトピックはデフォルトで作成されます。このトピックにアクセスするコネクターには、以下の ACL を構成する必要があります。

コネクターでトピックの作成が必要な場合は、リソースクラスターでの CREATE および DESCRIBE。
_confluent-command トピックでの DESCRIBE、READ、および WRITE。

ライセンスを使用する各プリンシパルに個々にアクセス許可を付与することも、ワイルドカード入力を使用してすべてのクライアントを許可することもできます。以下の例は、リソースクラスターおよび _confluent-command トピックの ACL を構成するために使用できるコマンドを示しています。

リソースクラスターの CREATE および DESCRIBE の ACL を設定します。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
--add --allow-principal User:<principal> \
--operation CREATE --operation DESCRIBE --cluster

_confluent-command トピックに対する DESCRIBE、READ、および WRITE の ACL を設定します。

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf \
--add --allow-principal User:<principal> \
--operation DESCRIBE --operation READ --operation WRITE --topic _confluent-command

デフォルトの構成プロパティのオーバーライド¶

confluent.topic.replication.factor を使用することにより、レプリケーション係数をオーバーライドできます。たとえば、（開発およびテスト用の）ブローカー数 3 未満の環境で送信先として Kafka クラスターを使用する場合、confluent.topic.replication.factor プロパティに 1 を設定する必要があります。

プロデューサー固有のプロパティは、confluent.topic.producer. プレフィックスを使用してオーバーライドできます。コンシューマー固有のプロパティは、confluent.topic.consumer. プレフィックスを使用してオーバーライドできます。

デフォルト値を使用することも、他のプロパティをカスタマイズすることもできます。たとえば、confluent.topic.client.id プロパティのデフォルトは、コネクターの名前に -licensing サフィックスを付けたものです。クライアント接続に SSL または SASL を必要とするブローカーの構成設定では、このプレフィックスを指定できます。

トピックのクリーンアップのポリシーはオーバーライドできません。トピックは、常に単一パーティションであり、圧縮されるからです。また、このプレフィックスを使用してシリアライザーおよびデシリアライザーを指定しないでください。追加しても無視されます。