gRPCコネクタ

概要

gRPCコネクタは、クライアントリーダー（ポーリング型ユーナリ呼び出し）およびライター（ペイロードをgRPCリクエストとして転送）の両方としてgRPCサービスを呼び出すことを可能にします。内部でHTTP/2を使用し、オプションのTLSをサポートするため、モダンなマイクロサービス、エッジゲートウェイ、gRPCエンドポイントを公開する産業APIとの統合に適しています。

コネクタタイプ:

• GrpcReader - 定期的にユーナリgRPCメソッドを呼び出し、レスポンスをMeddleペイロードとして発行
• GrpcWriter - 各受信ペイロードをgRPCリクエストとしてターゲットサービスに送信

機能

• ✅ オプションのTLS暗号化を伴うHTTP/2トランスポート
• ✅ 設定可能なレートでのポーリングベースのユーナリリーダー
• ✅ ライターはペイロードをJSONエンコードされたgRPCリクエストとして転送
• ✅ 読み取りポーリング用の静的リクエストボディ
• ✅ ストリーミング対応の設定フラグ
• ✅ 自動接続管理と再利用
• ✅ 安全のための呼び出しごとの10秒タイムアウト

基本設定

gRPC Reader

プレーンテキスト

{
  "type": "GrpcReader",
  "config": {
    "address": "192.168.1.50:50051",
    "service": "iot.SensorService",
    "method": "GetReading",
    "pollingRate": 1000,
    "requestBody": "{\"sensorId\":\"sensor-1\"}"
  }
}

TLSあり

{
  "type": "GrpcReader",
  "config": {
    "address": "edge.example.com:443",
    "service": "iot.SensorService",
    "method": "GetReading",
    "tls": true,
    "pollingRate": 5000,
    "requestBody": "{\"sensorId\":\"sensor-1\"}"
  }
}

ストリーミング

{
  "type": "GrpcReader",
  "config": {
    "address": "192.168.1.50:50051",
    "service": "iot.TelemetryService",
    "method": "Subscribe",
    "streaming": true,
    "pollingRate": 1000,
    "requestBody": "{\"deviceId\":\"plc-line-1\"}"
  }
}

gRPC Writer

{
  "type": "GrpcWriter",
  "config": {
    "address": "192.168.1.50:50051",
    "service": "iot.SensorService",
    "method": "PushReading",
    "tls": false
  }
}

設定パラメータ

Address

gRPCサーバーのホストとポート。

{
  "address": "192.168.1.50:50051"
}

形式: host:port（スキームプレフィックスなし）。TLSエンドポイントの場合は、アドレスにエンコードするのではなくtls: trueを設定してください。

Service

.protoファイルで定義された、パッケージプレフィックスを含む完全修飾されたgRPCサービス名。

{
  "service": "iot.SensorService"
}

形式: package.ServiceName。コネクタは内部で完全なメソッドパスを/{service}/{method}として構成します。

Method

呼び出すサービス上のメソッド名。

{
  "method": "GetReading"
}

protoの定義で宣言された通りのメソッド名と正確に一致する必要があります（大文字小文字を区別）。

TLS

接続のTLSを有効にします。無効の場合、コネクタは安全でない（プレーンテキスト）資格情報を使用します。

{
  "tls": true
}

推奨値:

• false - ネットワーク内の信頼できる環境（LAN/PLC）
• true - 公開エンドポイント、クラウドサービス、信頼できないネットワークを越えるもの

Request Body

ポーリングごとにリクエストペイロードとして送信される静的JSONボディ。

{
  "requestBody": "{\"sensorId\":\"sensor-1\",\"includeHistory\":false}"
}

注意:

• 有効なJSON文字列である必要があります。内部の引用符は示されているようにエスケープしてください
• 省略した場合、空のJSONオブジェクト{}が送信されます
• リクエストボディはjson.RawMessageとして扱われ、ポーリングごとにそのまま転送されます

Polling Rate

連続するgRPC呼び出し間のミリ秒単位の間隔（リーダーのみ）。

{
  "pollingRate": 1000
}

推奨値:

• 高速: 100-500ms
• 通常: 1000ms（1秒）
• 低速: 5000ms以上（5秒以上）

Streaming

ターゲットメソッドがサーバーストリーミングセマンティクスを使用することを示すフラグ。

{
  "streaming": true
}

設定されている場合、コネクタは依然としてポーリングのケイデンスで呼び出しを発行しますが、ダウンストリームのツールが期待を調整できるように、設定をストリーミング対応としてタグ付けします。徐々に大きくなるペイロードを返すメソッドや、より長い接続ライフサイクルを維持するメソッドに使用します。

データフロー

リーダーデータフロー

gRPCサーバー → GrpcReader（ユーナリ呼び出し） → JSONレスポンス → Meddleペイロード

例:

リーダー設定:

{
  "address": "192.168.1.50:50051",
  "service": "iot.SensorService",
  "method": "GetReading",
  "requestBody": "{\"sensorId\":\"sensor-1\"}",
  "pollingRate": 1000
}

サーバーレスポンス:

{
  "temperature": 24.7,
  "humidity": 58.3,
  "battery": 0.91,
  "timestamp": "2026-05-20T10:14:33Z"
}

発行されるMeddleペイロード:

{
  "temperature": 24.7,
  "humidity": 58.3,
  "battery": 0.91,
  "timestamp": "2026-05-20T10:14:33Z"
}

ライターデータフロー

Meddleペイロード → GrpcWriter（JSONエンコード） → gRPCサーバー

例:

受信ペイロード:

{
  "deviceId": "plc-line-1",
  "setpoint": 75.0,
  "mode": "auto"
}

ライターはペイロードをJSONエンコードし、それをリクエストボディとして/iot.SensorService/PushReadingを呼び出します。サーバーからのレスポンスは読み取られて破棄されます。

一般的なユースケース

1. マイクロサービステレメトリAPIのポーリング

LAN経由で内部マイクロサービスからセンサーデータを読み取る：

{
  "type": "GrpcReader",
  "config": {
    "address": "telemetry.internal:50051",
    "service": "telemetry.DeviceService",
    "method": "GetMetrics",
    "pollingRate": 2000,
    "requestBody": "{\"deviceId\":\"compressor-3\"}"
  }
}

2. 集約データをクラウドgRPCエンドポイントへプッシュ

処理されたペイロードをTLS経由でクラウドサービスに転送：

{
  "type": "GrpcWriter",
  "config": {
    "address": "api.example.cloud:443",
    "service": "ingest.MetricsService",
    "method": "Push",
    "tls": true
  }
}

3. Meddleインスタンス間のエッジ間通信

gRPCをトランスポートとして使用して2つのMeddleデプロイメントをブリッジ：

{
  "type": "GrpcWriter",
  "config": {
    "address": "edge-aggregator.local:50052",
    "service": "meddle.BridgeService",
    "method": "Forward",
    "tls": false
  }
}

トラブルシューティング

接続拒否

問題: connection refusedまたはtransport: error while dialing

解決策:

1. アドレスとポートが正しいことを確認
2. gRPCサーバーが実行されてリッスンしていることを確認
3. ファイアウォールルールを確認（telnet host portで素早く到達可能性をテスト）
4. サーバーがTLSを必要とする場合はtls: trueを設定

メソッドが見つからない

問題: rpc error: code = Unimplementedまたはunknown method

解決策:

1. serviceがパッケージを含む完全修飾名であることを確認（例：SensorServiceではなくiot.SensorService）
2. methodがprotoの宣言と正確に一致することを確認（大文字小文字を区別）
3. grpcurlのようなツールを使ってエンドポイント上のサービスをリスト：

   grpcurl -plaintext 192.168.1.50:50051 list

無効なリクエストボディ

問題: サーバー上でのデシリアル化エラー

解決策:

1. requestBodyが整形式のJSONであることを検証
2. フィールド名がprotoのフィールド名と正確に一致することを確認（gRPC-JSONトランスコーディングは大文字小文字を区別）
3. サーバーがprotobufバイナリを期待する場合、サーバー側でgRPC-JSONトランスコーディング（またはgrpc-gateway）が有効になっている必要があります

TLSハンドシェイク失敗

問題: TLSハンドシェイクまたは証明書エラー

解決策:

1. サーバー証明書が有効で信頼されていることを確認
2. アドレスのホスト名が証明書のSANと一致することを確認
3. 自己署名証明書の場合、CAをMeddleホストのトラストストアにインストール

タイムアウト

問題: すべての呼び出しで繰り返される10秒タイムアウト

解決策:

1. サーバー側のレイテンシを調査
2. レスポンスが一貫して遅い場合、ユーナリ10秒タイムアウトは厳格な上限です — ストリーミングメソッドまたは別のトランスポートを検討してください
3. 遅いレスポンスの背後で呼び出しがスタックしないようpollingRateを減らす

ベストプラクティス

1. 信頼できるネットワーク外では常にTLSを使用

公開インターネットを横断するエンドポイントではtls: falseのままにしないでください：

{
  "tls": true
}

2. ポーリングレートは正直に保つ

リモートgRPCサーバーに対する100msのポーリングはダウンストリームのレート制限を飽和させる可能性があります。1000msで始め、正当化される場合のみ厳しくします。

3. 安定した、よく知られたサービスを使用

明示的な.proto契約とバージョン化されたパッケージ（例：iot.v1.SensorService）を持つサービスを優先し、サーバーアップグレード時にリーダー設定が壊れないようにします。

4. リクエストボディを意図的にエンコード

インラインrequestBodyは静的パラメータには最適ですが、動的な値には適していません。動的なリクエストボディには、リーダーの代わりに上流のReshapeまたはTransformコネクタをGrpcWriterと組み合わせます。

5. Validationと組み合わせる

サーバースキーマはMeddleパイプラインよりも速く進化する可能性があるため、ダウンストリームのライターの前に常にgRPCレスポンスをValidationコネクタを通します。

ワークフロー例

リーダー → ストレージパイプライン

GrpcReader → Validation → Reshape → InfluxDb2Writer

1. GrpcReader: 1秒ごとに/iot.SensorService/GetReadingをポーリング
2. Validation: 必要なキーが存在し数値であることを保証
3. Reshape: location、device_typeなどのタグを追加
4. InfluxDb2Writer: 時系列データベースに保存

PLC → クラウドブリッジ

ModbusReader → Filter → GrpcWriter（クラウド取り込み）

1. ModbusReader: 1秒のケイデンスでPLCからレジスタ値を取得
2. Filter: 転送するキーのみを保持
3. GrpcWriter: フィルタリングされたペイロードをTLS経由でクラウドのingest.MetricsService.Pushメソッドにプッシュ

関連コネクタ

• HTTP Client - REST/HTTPサービスの代替
• MQTTv5 - 非同期ブローカーベースの統合
• Kafka - 高スループットストリーミングの代替
• Validation - gRPCレスポンスペイロードを検証
• Reshape - gRPCレスポンスからのフィールドのリネームと拡充

追加リソース

• gRPC公式サイト
• gRPCプロトコルドキュメント
• Protocol Buffersリファレンス
• grpcurl - エンドポイントテスト用のコマンドラインgRPCクライアント
• gRPCステータスコード