1 of 5

API Reference

This page describes the Lenses API references.

Lenses offers a set of REST and WebSocket APIs to help you provision and manage the application, make use of its capabilities, and query data via SQL.

Authenticate

Learn how to authenticate against Lenses APIs.

Lenses API Spec

Explore the Lenses API Spec.

Websocket Spec

Lenses uses websockets for SQL queries. Learn how to use Lenses Websockets.

Provisioning API

Explore the Provisioning API Spec.

Authentication API

This page describes how to authentication against Lenses APIs.

All requests must be authenticated using an HTTP Header x-kafka-lenses-token:myToken. You can obtain the token via the following login API or you can use a service account.

All REST APIs are protected via role-based authentication that is either BASIC or LDAP based, depending on how Lenses security has been set up. In order to be able to use the APIs, you will need to first authenticate via an appropriate user, then receive an access token and use that token for any subsequent request.

POST/api/login

Headers

Name

Value

Content-Type

application/json

Body

Name

Type

Description

user

string

Name of the user

password

string

Password of the user

Response

"a1f44cb8-0f37-4b96-828c-57bbd8d4934b"

UNAUTHORIZED

To run the example below, we recommend installing the jq tool

Example Request

 # login and receive the access token
HOST="http://localhost:3030"
curl -X POST -H "Content-Type:application/json" \
-d '{"user":"admin",  "password":"admin"}' \
${HOST}/api/login -s --compressed -w '\n'

Example Response

"a1f44cb8-0f37-4b96-828c-57bbd8d4934b"

Lenses API Spec

API reference documentation for Lenses.

https://api.lenses.io/

Websocket Spec

Data Access - Query Data via SQL

A WebSocket allows the client to request data using SQL as input.

Use the following URL to open a WebSocket connection. Depending on deployment, use ws (not-secure) or wss (secure) connection:

WS | WSS ${LENSES_URL}/api/ws/v2/sql/execute

Once the connection is established, the server expects the client to send a JSON message with the following structure:

{
  "token": String,
  "sql": String,
  "live": Optional[Boolean],
  "stats": Optional[Int]
}

Field

Description

Type

token

long

sql

The SQL query used to fetch data

string

live

If set, enables partial results for aggregation queries.

optional[boolean]

stats

Millis interval to receive query stats. If not provided the stats information is not sent

optional[int]

The client should wait for incoming messages and stop when EOF is received. The server sends JSON-encoded messages, the payload structure is:

{
  "type": <Enumeration>,
  "data": <Dependent on the type field>
}

Since there are multiple types of messages the server sends, the type attribute determines the data structure for the data payload. Below is the list of possible type values:

Value

Description

HEARTBEAT

To keep the connection alive, the server injects an empty message.

RECORD

The message represents a data record.

ERROR

The message contains information about an error that occurred.

STATS

The message contains information about the current execution.Only activated if the stats field in the request is present.

METADATA

The message contains a list of metadata fields each record message will contain. For example: offset, partition,timestamp, __keysize, __valueSize, valueSchemaId.

SCHEMA

Reserved.

PAGE_END

Reserved.

END

The message signals the end of the execution. The server will close the socket after this message is sent.

BADRECORD

The message contains information about the Kafka message which cannot be read. For example, if the topic expects Avro and the payload is not a valid Avro.

SENTINEL

The message contains information about the execution termination reason when query thresholds are reached.

Message types

Heartbeat

The data the attribute is null. The client should ignore these messages.

{
  "type": "HEARTBEAT"
}

Error

In this case, the payload structure is:

{
  "type": "HEARTBEAT",
  "data": String
}

Stats

When requested the payload format is:

{
  "type": "HEARTBEAT",
  "data":   {
    id:              String,
    bytesRead:       Long,
    sources:         Map[UUID, String],
    sourceConfigs:   Map[UUID, Map[String, Any]],
    sourcesStats:    Map[UUID, Map[Int, Map[String, Long]]],
    badRecordsTotal: Long,
    recordsSkipped:  Long,
    results:         Long,
    recordsScanned:  Long,
    duration:        Long,
    startedAt:       Long,
}

A possible payload can be:

{
  "type": "STATS",
  "data": {
    "id": "6a29ef99-3914-4640-b0dc-ec4a07c7ffc5",
    "bytesRead": 51972426,
    "sources": {
      "ad1afc73-3dec-41e7-b679-aa35c55e3328": "cc_payments"
    },
    "sourceConfigs": {
      "ad1afc73-3dec-41e7-b679-aa35c55e3328": {
        "live.aggs": true,
        "max.idle.time": 20000,
        "format.timestamp": true,
        "show.bad.records": true,
        "kafka.offset.timeout": 10000,
        "max.query.time": 3600000,
        "max.size": 209715200
      }
    },
    "sourcesStats": {
      "ad1afc73-3dec-41e7-b679-aa35c55e3328": {
        "0": {
          "begin": 0,
          "offsetBound": 541950,
          "bytesRead": 51972426,
          "recordsScanned": 541951,
          "end": 541950
        }
      }
    },
    "badRecordsTotal": 0,
    "recordsSkipped": 0,
    "results": 119,
    "recordsScanned": 541951,
    "duration": 18196,
    "startedAt": 1678901355694
  }
}

Metadata

The payload format is:

{
  "type": "METADATA",
  "data": {
    "fields": Array[String]
  }
}

End

The payload format is:

{
  "type": "END"
}

Bad Record

The payload format is:

{
  "type": "BADRECORD",
  "data": {
    "metadata": {
      "offset": Optional[Long],
      "partition": Optional[Long],
      "timestamp": Optional[Long],
      "table": String,
      "sourceId": String
    }
  }
}

Sentinel

The payload format is:

{
  "type": "SENTINEL",
  "data": {
    "event": String,
    "reason": {
      "type": String,
      "timeout":  Optional[Long],
      "source":   String,
      "sourceId": String,
      "amount":Long        
    }
  }
}

Record

Given the type RECORD the payloads contains the data returned to the client, and it is dependent on the SQL input. The payload format is:

{
  "type": "RECORD",
  "data": {
    "fields": <dynamic>,
    "value": <dynamic>,
    "metadata": <dynamic>,
    "rownum": Optional[Long]
  }
}

Value

Description

data.key

Returns the underlying Kafka message key value. Only applied when * projection is used. When a key projection is used (i.e. _key.MMSI), the value is returned as data.value.MMSI.

data.data

Will contain the output generated by the SQL projections. If * is used for the SQL projection, it will return the Kafka message value structure.

data.metadata

Returned when no aggregations are involved. It contains the Kafka message partition, offset, byte size information

data.rownum

Reserved optional long value.

Examples

For a topic which contains nested data for both key and value.

SELECT * FROM sea_vessel_position_reports

The payload format is:

{
  "type": "RECORD",
  "data": {
    "key": {
      "MMSI": 219005662
    },
    "value": {
      "Type": 1,
      "Repeat": 0,
      "MMSI": 219005662,
      "Speed": 0.1,
      "Accuracy": false,
      "Longitude": 12.570543333333333,
      "Latitude": 55.85040166666667,
      "location": "55.850402,12.570543",
      "Course": 177.9,
      "Heading": 511,
      "Second": 20,
      "RAIM": false,
      "Radio": 33577,
      "Status": 3,
      "Turn": -128.0,
      "Maneuver": 0,
      "Timestamp": 1491318144456507320
    },
    "metadata": {
      "offset": 71,
      "partition": 1,
      "timestamp": "2023-03-13T11:42:24.267Z",
      "__keysize": 10,
      "__valuesize": 79,
      "keySchemaId": 4,
      "valueSchemaId": 5
    },
    "rownum": 71
  }
}

If the Kafka message key is null the response would be:

{
  "type": "RECORD",
  "data": {
    "value": {
      "Type": 1,
      "Repeat": 0,
      "MMSI": 219005662,
      "Speed": 0.1,
      "Accuracy": false,
      "Longitude": 12.570543333333333,
      "Latitude": 55.85040166666667,
      "location": "55.850402,12.570543",
      "Course": 177.9,
      "Heading": 511,
      "Second": 20,
      "RAIM": false,
      "Radio": 33577,
      "Status": 3,
      "Turn": -128.0,
      "Maneuver": 0,
      "Timestamp": 1491318144456507320
    },
    "metadata": {
      "offset": 71,
      "partition": 1,
      "timestamp": "2023-03-13T11:42:24.267Z",
      "__keysize": 10,
      "__valuesize": 79,
      "keySchemaId": 4,
      "valueSchemaId": 5
    },
    "rownum": 71
  }
}

Projecting the key will not return the key attribute:

SELECT _key.MMSI FROM sea_vessel_position_reports

produces the following output:

{
  "type": "RECORD",
  "data": {
    "value": {
      "MMSI": 265527470
    },
    "metadata": {
      "offset": 86,
      "partition": 1,
      "timestamp": "2023-03-13T11:42:29.107Z",
      "__keysize": 10,
      "__valuesize": 79,
      "keySchemaId": 4,
      "valueSchemaId": 5
    },
    "rownum": 86
  }
}

Aggregating data will not return the key and metadata attributes:

SELECT count(*), currency 
FROM cc_payments
GROUP BY currency

{
  "type": "RECORD",
  "data": {
    "value": {
      "count": 20066,
      "currency": "NOR"
    },
    "rownum": 90
  },
  "rowId": "fc767952-97e5-46cf-868c-7cbfbcde21fb",
  "statementIndex": null
}

Provisioning API Spec

Update Lenses connections state.

put

It will update the connections state and validate the configuration. If the validation fails, the state will not be updated.

Query parameters

validateOnlybooleanoptional

It will only validate the request, not applying any actual change to the system.

Default: false

validateConnectivitybooleanoptional

It will try to connect to the configured service as part of the validation step.

Default: true

Body

provisioningobjectoptional

Configuration in YAML format representing the connections state.

attachedFilestring · binaryoptional

Attached file(s) needed for establishing the connection. The name of each file part is used as a reference in the manifest.

Responses

application/json

cURL

JavaScript

Python

HTTP

curl -L \
  --request PUT \
  --url '/api/v1/state/connections' \
  --header 'Content-Type: multipart/form-data' \
  --form 'provisioning=[object Object]' \
  --form 'attachedFile=binary'

200

400

{
  "updated": [
    "text"
  ],
  "created": [
    "text"
  ],
  "deleted": [
    "text"
  ]
}

Successfully updated connection state

Retrieve system state

get

Responses

application/json

cURL

JavaScript

Python

HTTP

curl -L \
  --url '/api/v1/state'

200

{
  "license": {
    "maxBrokers": 1,
    "expiry": 1,
    "clientId": "text",
    "isRespected": true,
    "status": "Valid",
    "message": "text"
  },
  "connections": {
    "kafka": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "confluentSchemaRegistry": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "elasticSearch": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "pagerDuty": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "datadog": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "slack": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "alertManager": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "webhook": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "aws": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "connect": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "awsGlueSchemaRegistry": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "zookeeper": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "postgres": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "splunk": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ],
    "kerberos": [
      {
        "name": "text",
        "version": 1,
        "tags": [
          "text"
        ]
      }
    ]
  }
}

Successful retrieval of system state

Update the license data

put

Body

sourcestringoptional

clientIdstringoptional

detailsstringoptional

keystringoptional

Responses

application/json

cURL

JavaScript

Python

HTTP

curl -L \
  --request PUT \
  --url '/api/v1/state/license' \
  --header 'Content-Type: application/json' \
  --data '{
    "source": "text",
    "clientId": "text",
    "details": "text",
    "key": "text"
  }'

200

400

{
  "maxBrokers": 1,
  "expiry": 1,
  "clientId": "text",
  "isRespected": true,
  "status": "Valid",
  "message": "text",
  "settings": {
    "security": {
      "root": {
        "enabled": true
      },
      "basic": {
        "enabled": true
      },
      "ldap": {
        "enabled": true
      },
      "kerberos": {
        "enabled": true
      },
      "custom": {
        "enabled": true
      },
      "sso": {
        "enabled": true
      },
      "serviceAccount": {
        "enabled": true,
        "restriction": {
          "name": "None"
        }
      }
    },
    "sql": {
      "streaming": {
        "enabled": true,
        "restriction": {
          "name": "None"
        }
      },
      "sql": {
        "enabled": true
      }
    },
    "kafkaSettings": {
      "acls": true,
      "quotas": true,
      "consumerOffsetManagement": true
    },
    "audit": {
      "enabled": true,
      "integration": true
    },
    "connections": {
      "enabled": true
    },
    "application": {
      "topology": true,
      "connectorsOnKubernetes": true
    },
    "approval": {
      "enabled": true
    },
    "alerts": {
      "enabled": true,
      "rules": {
        "name": "None"
      },
      "integration": {
        "enabled": true,
        "channels": [
          "text"
        ],
        "max": {
          "name": "None"
        }
      }
    },
    "data": {
      "masking": true,
      "customSerde": true,
      "sla": true,
      "namespace": {
        "enabled": true,
        "max": {
          "name": "None"
        }
      }
    },
    "backup": {
      "enabled": true
    }
  },
  "currentTime": 1
}

License successfully updated and current license info returned

API Reference

Authentication API

Example Request

Example Response

Lenses API Spec

Websocket Spec

Data Access - Query Data via SQL

Message types

Heartbeat

Error

Stats

Metadata

End

Bad Record

Sentinel

Record

Examples

Provisioning API Spec

Update Lenses connections state.

Show child attributes

Retrieve system state

Update the license data