From Launching an Amazon Timestream for InfluxDB Instance to Using It in an Application (InfluxDB 3, Single Instance)
This page has been translated by machine translation. View original
Introduction
I tried using Amazon Timestream for LiveAnalytics (the former Timestream) in a project, and my interest was piqued, so I decided to try out the InfluxDB version (the new Timestream) as well.
※ It has been announced that Amazon Timestream for LiveAnalytics (the former Timestream) will no longer be available for new deployments as of June 20, 2025.
Roadmap
- Launch from the managed console
- CDK implementation
- Build a sample app
- Connectivity verification
Launching from the Managed Console
-
On the Timestream > InfluxDB databases page, click Create InfluxDB database.
-
For Engine settings, select Engine version = [InfluxDB 3] and InfluxDB Edition = [Core (Standard Workloads & Dev/Test)] as shown in the attachment, and set InfluxDBV3CoreMediumDefault for the parameter group.
-
Set a DB name of your choice and create the instance with medium size. Use the managed key.
-
Leave the remaining settings as they are and click Create InfluxDB.
※ Note: The VPC in which you place a private Timestream for InfluxDB must have an S3 endpoint associated with it.
This allowed us to launch a new Timestream for InfluxDB. (I'm not sure whether I should hide it, but I've masked the ID just in case.)
CDK Implementation
We will implement the Timestream for InfluxDB built in the management console so that it can be reproduced using AWS CDK (TypeScript).
Project Initialization
First, create the CDK project. Since I want to use pnpm, I'll delete node_modules and package-lock.json and reinstall with pnpm.
cdk init app --language typescript
rm -rf node_modules package-lock.json
pnpm install
pnpm update aws-cdk --latest
CDK Support Status for Timestream for InfluxDB
Currently, only L1 constructs (CfnInfluxDBInstance, CfnInfluxDBCluster) are provided for Timestream for InfluxDB in the aws-timestream module of aws-cdk-lib (confirmed with aws-cdk-lib 2.265.0). L2 constructs are not yet available, so you need to use L1 directly. There is also no CloudFormation resource yet for creating parameter groups themselves, and the related issue in the CDK repository has a needs-cfn label (waiting for support on the CloudFormation side).
Preparing the Network and Secrets
As noted when building in the management console, private subnets require an S3 gateway endpoint. Create a new VPC with CDK and also create the endpoint along with it.
Since I don't want to write the initial password in plain text, I'll auto-generate it using Secrets Manager and inject it into the instance via CloudFormation dynamic references.
const vpc = new ec2.Vpc(this, 'InfluxVpc', {
maxAzs: 2,
natGateways: 0,
subnetConfiguration: [
{ name: 'influx-private', subnetType: ec2.SubnetType.PRIVATE_ISOLATED, cidrMask: 24 },
],
});
vpc.addGatewayEndpoint('S3Endpoint', {
service: ec2.GatewayVpcEndpointAwsService.S3,
});
const securityGroup = new ec2.SecurityGroup(this, 'InfluxSecurityGroup', {
vpc,
description: 'Allow InfluxDB client access from within the VPC',
allowAllOutbound: true,
});
securityGroup.addIngressRule(
ec2.Peer.ipv4(vpc.vpcCidrBlock),
ec2.Port.tcp(8086),
'InfluxDB API access from within the VPC',
);
const adminPassword = new secretsmanager.Secret(this, 'InfluxAdminPassword', {
generateSecretString: { excludePunctuation: true, passwordLength: 24 },
});
About Specifying the Parameter Group
CfnInfluxDBInstance has a property called dbParameterGroupIdentifier, and it should be possible to explicitly associate a parameter group such as InfluxDBV3CoreMediumDefault shown in the management console screen. However, there is no CloudFormation resource yet for creating a parameter group for Timestream for InfluxDB (the same response has been given from the AWS side in aws/aws-cdk#31862).
In fact, when I tried deploying with InfluxDBV3CoreMediumDefault explicitly specified in dbParameterGroupIdentifier exactly as shown in the management console, it intermittently resulted in CREATE_FAILED with the error "The parameter group with id does not exist". Since the same value sometimes succeeded and sometimes failed and I could not clearly identify the cause, I've decided to omit the dbParameterGroupIdentifier specification this time. The AWS official documentation also states that "if you create a DB instance without specifying a DB parameter group, the InfluxDB engine defaults are used," and with this approach, instances can be created stably.
Final Stack
import * as cdk from 'aws-cdk-lib/core';
import * as ec2 from 'aws-cdk-lib/aws-ec2';
import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager';
import * as timestream from 'aws-cdk-lib/aws-timestream';
import { Construct } from 'constructs';
export class TimestreamInfluxDemoStack extends cdk.Stack {
constructor(scope: Construct, id: string, props?: cdk.StackProps) {
super(scope, id, props);
const vpc = new ec2.Vpc(this, 'InfluxVpc', {
maxAzs: 2,
natGateways: 0,
subnetConfiguration: [
{ name: 'influx-private', subnetType: ec2.SubnetType.PRIVATE_ISOLATED, cidrMask: 24 },
],
});
vpc.addGatewayEndpoint('S3Endpoint', {
service: ec2.GatewayVpcEndpointAwsService.S3,
});
const securityGroup = new ec2.SecurityGroup(this, 'InfluxSecurityGroup', {
vpc,
description: 'Allow InfluxDB client access from within the VPC',
allowAllOutbound: true,
});
securityGroup.addIngressRule(
ec2.Peer.ipv4(vpc.vpcCidrBlock),
ec2.Port.tcp(8086),
'InfluxDB API access from within the VPC',
);
const adminPassword = new secretsmanager.Secret(this, 'InfluxAdminPassword', {
generateSecretString: { excludePunctuation: true, passwordLength: 24 },
});
const instance = new timestream.CfnInfluxDBInstance(this, 'InfluxInstance', {
name: 'db-sample-cdk',
username: 'admin',
password: adminPassword.secretValue.unsafeUnwrap(),
organization: 'test',
bucket: 'demo',
dbInstanceType: 'db.influx.medium',
dbStorageType: 'InfluxIOIncludedT1',
allocatedStorage: 20,
deploymentType: 'SINGLE_AZ',
networkType: 'IPV4',
publiclyAccessible: false,
vpcSecurityGroupIds: [securityGroup.securityGroupId],
vpcSubnetIds: vpc.isolatedSubnets.map((subnet) => subnet.subnetId),
});
new cdk.CfnOutput(this, 'InfluxEndpoint', { value: instance.attrEndpoint });
new cdk.CfnOutput(this, 'InfluxAdminPasswordSecretArn', { value: adminPassword.secretArn });
}
}
Running cdk deploy, the DB instance was successfully created.
Building a Sample App
The third item on the roadmap: building a sample app. The configuration is simple, consisting of the following two components.
- edge: A container simulating an IoT device. Generates dummy temperature and humidity data every 5 seconds and writes it to InfluxDB.
- viewer: A web app that aggregates the written data and displays it as a graph. Accessible from a browser via ALB.
Both run on ECS Fargate.
edge: Writing Sensor Data
const ENDPOINT = process.env.INFLUX_ENDPOINT as string;
const BUCKET = process.env.INFLUX_BUCKET as string;
const ORGANIZATION = process.env.INFLUX_ORG as string;
const TOKEN = process.env.INFLUX_TOKEN as string;
const DEVICE_ID = process.env.DEVICE_ID || 'edge-01';
const INTERVAL_MS = Number(process.env.WRITE_INTERVAL_MS || 5000);
function randomInRange(min: number, max: number): number {
return Math.round((min + Math.random() * (max - min)) * 10) / 10;
}
async function writePoint(): Promise<void> {
const temperature = randomInRange(18, 28);
const humidity = randomInRange(30, 70);
const timestamp = Math.floor(Date.now() / 1000);
const line = `sensor,device_id=${DEVICE_ID} temperature=${temperature},humidity=${humidity} ${timestamp}`;
const url = `https://${ENDPOINT}:8086/api/v2/write?org=${encodeURIComponent(ORGANIZATION)}&bucket=${encodeURIComponent(BUCKET)}&precision=s`;
const res = await fetch(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${TOKEN}`,
'Content-Type': 'text/plain; charset=utf-8',
},
body: line,
});
const text = await res.text();
if (!res.ok || text.trimStart().startsWith('<')) {
console.error(`write failed: ${res.status} ${text.slice(0, 500)}`);
return;
}
console.log(`wrote point: ${line}`);
}
setInterval(() => {
writePoint().catch((err) => console.error('write error', err));
}, INTERVAL_MS);
Timestream for InfluxDB has two types of resources: CfnInfluxDBInstance (DB Instance, the single instance used this time) and CfnInfluxDBCluster (DB Cluster, for Enterprise multi-node configurations). Writing data to your Timestream for InfluxDB 3 cluster introduces the native Line Protocol write API (/api/v3/write_lp), but as the name suggests, this is documentation for DB Cluster. On the other hand, the DB Instance creation and connection tutorial Creating and connecting to a Timestream for InfluxDB instance only describes the procedure using Telegraf's InfluxDB v2-compatible output plugin (outputs.influxdb_v2) for writing, with no mention of the v3 native API.
In fact, when I sent a request to /api/v3/write_lp against the DB Instance (Core, single instance) created this time, only the HTML of the InfluxDB UI (SPA) was returned with status 200, and no data was written. If you only check the status code for success, you won't notice this condition.
viewer: Aggregating and Displaying Graphs
The viewer is a web app that aggregates data written to InfluxDB and displays it as a graph viewable from a browser. The same applies to the query side — when I sent requests to the v3 native SQL API (/api/v3/query_sql) or InfluxQL API (/api/v3/query_influxql) as described in Querying data from Timestream for InfluxDB 3, only the SPA HTML was returned, just like with the write API. On the other hand, the v2-compatible Flux query API (/api/v2/query) returned the expected results, so I'm using that this time.
import * as http from 'http';
const ENDPOINT = process.env.INFLUX_ENDPOINT as string;
const BUCKET = process.env.INFLUX_BUCKET as string;
const ORGANIZATION = process.env.INFLUX_ORG as string;
const TOKEN = process.env.INFLUX_TOKEN as string;
const PORT = Number(process.env.PORT || 8080);
const BASE_URL = `https://${ENDPOINT}:8086`;
const FLUX_QUERY = `
from(bucket: "${BUCKET}")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "sensor")
|> filter(fn: (r) => r._field == "temperature" or r._field == "humidity")
|> keep(columns: ["_time", "_value", "_field", "device_id"])
|> sort(columns: ["_time"])
`;
interface Point {
time: number; // epoch ms
value: number;
}
type SeriesByDevice = Record<string, Point[]>;
type SeriesByField = Record<string, SeriesByDevice>;
async function fetchSeries(): Promise<{ ok: true; data: SeriesByField } | { ok: false; status: number; body: string }> {
const url = `${BASE_URL}/api/v2/query?org=${encodeURIComponent(ORGANIZATION)}`;
const res = await fetch(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${TOKEN}`,
'Content-Type': 'application/vnd.flux',
Accept: 'application/csv',
},
body: FLUX_QUERY,
});
const text = await res.text();
if (!res.ok || text.trimStart().startsWith('<')) {
return { ok: false, status: res.status, body: text.slice(0, 2000) };
}
return { ok: true, data: parseAnnotatedCsv(text) };
}
function parseAnnotatedCsv(text: string): SeriesByField {
const lines = text.split('\n').map((l) => l.trimEnd());
const dataLines = lines.filter((l) => l.length > 0 && !l.startsWith('#'));
const result: SeriesByField = {};
if (dataLines.length === 0) return result;
const header = dataLines[0].split(',');
const timeIdx = header.indexOf('_time');
const valueIdx = header.indexOf('_value');
const fieldIdx = header.indexOf('_field');
const deviceIdx = header.indexOf('device_id');
if (timeIdx < 0 || valueIdx < 0 || fieldIdx < 0 || deviceIdx < 0) return result;
for (let i = 1; i < dataLines.length; i++) {
const line = dataLines[i];
if (line === dataLines[0]) continue;
const cols = line.split(',');
const time = Date.parse(cols[timeIdx]);
const value = Number(cols[valueIdx]);
const field = cols[fieldIdx];
const device = cols[deviceIdx];
if (!field || !device || Number.isNaN(time) || Number.isNaN(value)) continue;
if (!result[field]) result[field] = {};
if (!result[field][device]) result[field][device] = [];
result[field][device].push({ time, value });
}
return result;
}
// ...Continues with line graph rendering in SVG and HTTP server implementation
Since temperature and humidity have different units, rather than forcing them into a single graph, I display them as two separate line graphs with separate axes.
Issuing an API Token
To allow edge/viewer to write to and query InfluxDB, an API token is required separately from the username/password used during initial setup. Since I didn't want to paste it directly into the code, I configured the system to issue an API token via Lambda (custom resource) at deploy time and store it in Secrets Manager.
import {
SecretsManagerClient,
GetSecretValueCommand,
PutSecretValueCommand,
} from '@aws-sdk/client-secrets-manager';
const secretsClient = new SecretsManagerClient({});
export const handler = async (event: CloudFormationCustomResourceEvent) => {
if (event.RequestType === 'Delete') {
return { PhysicalResourceId: event.PhysicalResourceId };
}
const { Endpoint, Username, PasswordSecretArn, Organization: OrgName, Bucket: BucketName, ApiTokenSecretArn } =
event.ResourceProperties;
const passwordResp = await secretsClient.send(
new GetSecretValueCommand({ SecretId: PasswordSecretArn }),
);
const password = passwordResp.SecretString;
const baseUrl = `https://${Endpoint}:8086`;
// Start a session using the v2-compatible signin API
const signinResp = await fetch(`${baseUrl}/api/v2/signin`, {
method: 'POST',
headers: {
Authorization: `Basic ${Buffer.from(`${Username}:${password}`).toString('base64')}`,
},
});
const cookie = signinResp.headers.get('set-cookie');
const sessionCookie = cookie!.split(';')[0];
// Resolve orgID from organization name
const orgsResp = await fetch(`${baseUrl}/api/v2/orgs`, {
headers: { Cookie: sessionCookie },
});
const orgsBody = (await orgsResp.json()) as { orgs?: { id: string; name: string }[] };
const org = (orgsBody.orgs || []).find((o) => o.name === OrgName)!;
// Resolve bucket ID from bucket name. If resource.id is not specified,
// all resources of that type (all buckets under this organization) become the target
const bucketsResp = await fetch(`${baseUrl}/api/v2/buckets?org=${encodeURIComponent(OrgName)}`, {
headers: { Cookie: sessionCookie },
});
const bucketsBody = (await bucketsResp.json()) as { buckets?: { id: string; name: string }[] };
const bucket = (bucketsBody.buckets || []).find((b) => b.name === BucketName)!;
// Issue a token with only read/write permissions for the target bucket
const authResp = await fetch(`${baseUrl}/api/v2/authorizations`, {
method: 'POST',
headers: { Cookie: sessionCookie, 'Content-Type': 'application/json' },
body: JSON.stringify({
status: 'active',
description: 'timestream-influx-demo edge/viewer token',
orgID: org.id,
permissions: [
{ action: 'read', resource: { type: 'buckets', id: bucket.id, orgID: org.id } },
{ action: 'write', resource: { type: 'buckets', id: bucket.id, orgID: org.id } },
],
}),
});
const authBody = (await authResp.json()) as { token?: string };
await secretsClient.send(
new PutSecretValueCommand({ SecretId: ApiTokenSecretArn, SecretString: authBody.token }),
);
return { PhysicalResourceId: `token-provisioning-${BucketName}` };
};
I start a session using the v2-compatible signin API and resolve the orgID from the organization name. Since the InfluxDB v2-compatible authorization API targets the entire resource type (in this case, all buckets under the organization) if resource.id is omitted, I also resolve the bucket ID from the bucket name and issue a token with read/write permissions scoped to only that bucket.
Running on ECS Fargate
Both edge and viewer run on ECS Fargate. Since InfluxDB remains in the private subnet, I also added Interface endpoints (ECR, CloudWatch Logs, Secrets Manager) to the VPC for image retrieval, log delivery, and Secrets Manager access. The viewer's aggregated results are published via a public ALB so they can be viewed from a browser.
const vpc = new ec2.Vpc(this, 'InfluxVpc', {
maxAzs: 2,
natGateways: 0,
subnetConfiguration: [
{ name: 'influx-private', subnetType: ec2.SubnetType.PRIVATE_ISOLATED, cidrMask: 24 },
{ name: 'alb-public', subnetType: ec2.SubnetType.PUBLIC, cidrMask: 24 },
],
});
const interfaceEndpointSubnets = { subnetType: ec2.SubnetType.PRIVATE_ISOLATED };
vpc.addInterfaceEndpoint('EcrApiEndpoint', { service: ec2.InterfaceVpcEndpointAwsService.ECR, subnets: interfaceEndpointSubnets });
vpc.addInterfaceEndpoint('EcrDkrEndpoint', { service: ec2.InterfaceVpcEndpointAwsService.ECR_DOCKER, subnets: interfaceEndpointSubnets });
vpc.addInterfaceEndpoint('CloudWatchLogsEndpoint', { service: ec2.InterfaceVpcEndpointAwsService.CLOUDWATCH_LOGS, subnets: interfaceEndpointSubnets });
vpc.addInterfaceEndpoint('SecretsManagerEndpoint', { service: ec2.InterfaceVpcEndpointAwsService.SECRETS_MANAGER, subnets: interfaceEndpointSubnets });
// Match the CPU architecture of the Mac used for building with finch/Docker (arm64) to the task execution architecture
const runtimePlatform: ecs.RuntimePlatform = {
cpuArchitecture: ecs.CpuArchitecture.ARM64,
operatingSystemFamily: ecs.OperatingSystemFamily.LINUX,
};
const cluster = new ecs.Cluster(this, 'SampleAppCluster', { vpc });
const edgeTaskDefinition = new ecs.FargateTaskDefinition(this, 'EdgeTaskDefinition', { cpu: 256, memoryLimitMiB: 512, runtimePlatform });
edgeTaskDefinition.addContainer('EdgeContainer', {
image: ecs.ContainerImage.fromAsset(path.join(__dirname, '../apps/edge')),
logging: ecs.LogDrivers.awsLogs({ streamPrefix: 'edge' }),
environment: { INFLUX_ENDPOINT: instance.attrEndpoint, INFLUX_BUCKET: 'demo', INFLUX_ORG: 'test', DEVICE_ID: 'edge-01' },
secrets: { INFLUX_TOKEN: ecs.Secret.fromSecretsManager(apiTokenSecret) },
});
const viewerTaskDefinition = new ecs.FargateTaskDefinition(this, 'ViewerTaskDefinition', { cpu: 256, memoryLimitMiB: 512, runtimePlatform });
const viewerContainer = viewerTaskDefinition.addContainer('ViewerContainer', {
image: ecs.ContainerImage.fromAsset(path.join(__dirname, '../apps/viewer')),
logging: ecs.LogDrivers.awsLogs({ streamPrefix: 'viewer' }),
environment: { INFLUX_ENDPOINT: instance.attrEndpoint, INFLUX_BUCKET: 'demo', INFLUX_ORG: 'test', PORT: '8080' },
secrets: { INFLUX_TOKEN: ecs.Secret.fromSecretsManager(apiTokenSecret) },
});
viewerContainer.addPortMappings({ containerPort: 8080 });
const viewerService = new ecs.FargateService(this, 'ViewerService', {
cluster,
taskDefinition: viewerTaskDefinition,
desiredCount: 1,
vpcSubnets: { subnetType: ec2.SubnetType.PRIVATE_ISOLATED },
securityGroups: [viewerSecurityGroup],
assignPublicIp: false,
});
const viewerAlb = new elbv2.ApplicationLoadBalancer(this, 'ViewerAlb', {
vpc,
internetFacing: true,
securityGroup: albSecurityGroup,
vpcSubnets: { subnetType: ec2.SubnetType.PUBLIC },
});
const viewerListener = viewerAlb.addListener('ViewerListener', { port: 80, open: false });
viewerListener.addTargets('ViewerTarget', {
port: 8080,
protocol: elbv2.ApplicationProtocol.HTTP,
targets: [viewerService],
healthCheck: { path: '/health', interval: cdk.Duration.seconds(30) },
});
new cdk.CfnOutput(this, 'ViewerUrl', { value: `http://${viewerAlb.loadBalancerDnsName}` });
After deploying edge/viewer and the ALB with cdk deploy, the sensor data written by edge is now reflected in real time on the viewer's graph. Opening the ALB URL in a browser lets you check the temperature and humidity trends over the past hour as a graph.
Conclusion
- In this verification, there are some points described as "probably possible but doesn't seem to work based on what I observed." I will update these as primary information or updates become available going forward.