feat: MSK, MSK CONNECT and SES
This commit is contained in:
Zakaria
@@ -0,0 +1,273 @@
|
||||
import {
|
||||
CreateClusterCommand,
|
||||
DeleteClusterCommand,
|
||||
GetBootstrapBrokersCommand,
|
||||
KafkaClient,
|
||||
ListClustersCommand,
|
||||
type BrokerNodeGroupInfo,
|
||||
} from "@aws-sdk/client-kafka";
|
||||
import { Kafka } from "kafkajs";
|
||||
import { awsClients } from "../aws";
|
||||
|
||||
export type CreateMskClusterInput = {
|
||||
clusterName: string;
|
||||
kafkaVersion: string;
|
||||
numberOfBrokerNodes: number;
|
||||
brokerNodeGroupInfo: BrokerNodeGroupInfo;
|
||||
tags?: Record<string, string>;
|
||||
};
|
||||
|
||||
export type CreateTopicInput = {
|
||||
topic: string;
|
||||
numPartitions?: number;
|
||||
replicationFactor?: number;
|
||||
configEntries?: { name: string; value: string }[];
|
||||
};
|
||||
|
||||
export type ProduceMessageInput = {
|
||||
key?: string;
|
||||
value: string;
|
||||
headers?: Record<string, string>;
|
||||
};
|
||||
|
||||
type KafkaAdmin = {
|
||||
connect(): Promise<void>;
|
||||
disconnect(): Promise<void>;
|
||||
fetchTopicMetadata(input?: { topics?: string[] }): Promise<{
|
||||
topics: { name: string; partitions: unknown[] }[];
|
||||
}>;
|
||||
createTopics(input: { topics: CreateTopicInput[] }): Promise<boolean>;
|
||||
deleteTopics(input: { topics: string[] }): Promise<void>;
|
||||
fetchTopicOffsets(topic: string): Promise<{ partition: number; offset: string }[]>;
|
||||
fetchOffsets(input: {
|
||||
groupId: string;
|
||||
topics?: string[];
|
||||
}): Promise<{ topic: string; partitions: { partition: number; offset: string }[] }[]>;
|
||||
};
|
||||
|
||||
type KafkaProducer = {
|
||||
connect(): Promise<void>;
|
||||
disconnect(): Promise<void>;
|
||||
send(input: {
|
||||
topic: string;
|
||||
messages: { key?: string; value: string; headers?: Record<string, string> }[];
|
||||
}): Promise<{ topicName: string; partition: number; baseOffset?: string }[]>;
|
||||
};
|
||||
|
||||
type KafkaConsumer = {
|
||||
connect(): Promise<void>;
|
||||
disconnect(): Promise<void>;
|
||||
subscribe(input: { topic: string; fromBeginning: boolean }): Promise<void>;
|
||||
run(input: {
|
||||
autoCommit: boolean;
|
||||
eachMessage: (payload: {
|
||||
partition: number;
|
||||
message: {
|
||||
key: Buffer | null;
|
||||
value: Buffer | null;
|
||||
headers?: Record<string, Buffer>;
|
||||
offset: string;
|
||||
timestamp: string;
|
||||
};
|
||||
}) => Promise<void>;
|
||||
}): Promise<void>;
|
||||
stop(): Promise<void>;
|
||||
};
|
||||
|
||||
export type KafkaClientFactory = (brokers: string[]) => {
|
||||
admin(): KafkaAdmin;
|
||||
producer(): KafkaProducer;
|
||||
consumer(input: { groupId: string }): KafkaConsumer;
|
||||
};
|
||||
|
||||
function createKafkaClient(brokers: string[]) {
|
||||
return new Kafka({ brokers }) as unknown as ReturnType<KafkaClientFactory>;
|
||||
}
|
||||
|
||||
function splitBootstrapBrokers(bootstrapBrokers: string): string[] {
|
||||
return bootstrapBrokers.split(",").map((broker) => broker.trim()).filter(Boolean);
|
||||
}
|
||||
|
||||
export function createMskService(
|
||||
client: KafkaClient = awsClients.kafka,
|
||||
kafkaClientFactory: KafkaClientFactory = createKafkaClient,
|
||||
) {
|
||||
async function brokers(clusterArn: string): Promise<string[]> {
|
||||
const command = new GetBootstrapBrokersCommand({ ClusterArn: clusterArn });
|
||||
command.middlewareStack.add(
|
||||
(next) => async (args) => {
|
||||
const request = args.request as { headers: Record<string, string> };
|
||||
request.headers["x-floci-internal-client"] = "true";
|
||||
return next(args);
|
||||
},
|
||||
{ step: "build", name: "flociInternalMskClient" },
|
||||
);
|
||||
const response = await client.send(command);
|
||||
const bootstrapBrokers = response.BootstrapBrokerString ?? response.BootstrapBrokerStringTls;
|
||||
if (!bootstrapBrokers) {
|
||||
throw new Error(`MSK cluster ${clusterArn} did not return bootstrap brokers`);
|
||||
}
|
||||
return splitBootstrapBrokers(bootstrapBrokers);
|
||||
}
|
||||
|
||||
return {
|
||||
async listClusters() {
|
||||
const clusters = [];
|
||||
let nextToken: string | undefined;
|
||||
do {
|
||||
const response = await client.send(new ListClustersCommand({ NextToken: nextToken }));
|
||||
clusters.push(...(response.ClusterInfoList ?? []));
|
||||
nextToken = response.NextToken;
|
||||
} while (nextToken);
|
||||
return clusters;
|
||||
},
|
||||
|
||||
async createCluster(input: CreateMskClusterInput) {
|
||||
const response = await client.send(new CreateClusterCommand({
|
||||
ClusterName: input.clusterName,
|
||||
KafkaVersion: input.kafkaVersion,
|
||||
NumberOfBrokerNodes: input.numberOfBrokerNodes,
|
||||
BrokerNodeGroupInfo: input.brokerNodeGroupInfo,
|
||||
Tags: input.tags,
|
||||
}));
|
||||
return response;
|
||||
},
|
||||
|
||||
async deleteCluster(clusterArn: string) {
|
||||
return client.send(new DeleteClusterCommand({ ClusterArn: clusterArn }));
|
||||
},
|
||||
|
||||
async getBootstrapBrokers(clusterArn: string) {
|
||||
const response = await client.send(new GetBootstrapBrokersCommand({ ClusterArn: clusterArn }));
|
||||
return response.BootstrapBrokerString ?? response.BootstrapBrokerStringTls;
|
||||
},
|
||||
|
||||
async listTopics(clusterArn: string) {
|
||||
const admin = kafkaClientFactory(await brokers(clusterArn)).admin();
|
||||
await admin.connect();
|
||||
try {
|
||||
const metadata = await admin.fetchTopicMetadata();
|
||||
return metadata.topics
|
||||
.filter((topic) => !topic.name.startsWith("__"))
|
||||
.map((topic) => ({
|
||||
name: topic.name,
|
||||
partitionCount: topic.partitions.length,
|
||||
}));
|
||||
} finally {
|
||||
await admin.disconnect();
|
||||
}
|
||||
},
|
||||
|
||||
async createTopic(clusterArn: string, input: CreateTopicInput) {
|
||||
const admin = kafkaClientFactory(await brokers(clusterArn)).admin();
|
||||
await admin.connect();
|
||||
try {
|
||||
await admin.createTopics({ topics: [input] });
|
||||
} finally {
|
||||
await admin.disconnect();
|
||||
}
|
||||
},
|
||||
|
||||
async deleteTopic(clusterArn: string, topic: string) {
|
||||
const admin = kafkaClientFactory(await brokers(clusterArn)).admin();
|
||||
await admin.connect();
|
||||
try {
|
||||
await admin.deleteTopics({ topics: [topic] });
|
||||
} finally {
|
||||
await admin.disconnect();
|
||||
}
|
||||
},
|
||||
|
||||
async produceMessage(clusterArn: string, topic: string, input: ProduceMessageInput) {
|
||||
const producer = kafkaClientFactory(await brokers(clusterArn)).producer();
|
||||
await producer.connect();
|
||||
try {
|
||||
return await producer.send({ topic, messages: [input] });
|
||||
} finally {
|
||||
await producer.disconnect();
|
||||
}
|
||||
},
|
||||
|
||||
async peekMessages(clusterArn: string, topic: string, limit = 50) {
|
||||
const consumer = kafkaClientFactory(await brokers(clusterArn)).consumer({
|
||||
groupId: `floci-ui-peek-${crypto.randomUUID()}`,
|
||||
});
|
||||
const messages: { partition: number; offset: string; timestamp: string; key?: string; value?: string; headers: Record<string, string> }[] = [];
|
||||
await consumer.connect();
|
||||
try {
|
||||
await consumer.subscribe({ topic, fromBeginning: true });
|
||||
await new Promise<void>(async (resolve, reject) => {
|
||||
const timeout = setTimeout(resolve, 500);
|
||||
try {
|
||||
await consumer.run({
|
||||
autoCommit: false,
|
||||
eachMessage: async ({ partition, message }) => {
|
||||
if (messages.length >= limit) {
|
||||
return;
|
||||
}
|
||||
messages.push({
|
||||
partition,
|
||||
offset: message.offset,
|
||||
timestamp: message.timestamp,
|
||||
key: message.key?.toString(),
|
||||
value: message.value?.toString(),
|
||||
headers: Object.fromEntries(
|
||||
Object.entries(message.headers ?? {}).map(([name, value]) => [name, value.toString()]),
|
||||
),
|
||||
});
|
||||
if (messages.length >= limit) {
|
||||
clearTimeout(timeout);
|
||||
resolve();
|
||||
}
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
clearTimeout(timeout);
|
||||
reject(error);
|
||||
}
|
||||
});
|
||||
return messages;
|
||||
} finally {
|
||||
await consumer.stop();
|
||||
await consumer.disconnect();
|
||||
}
|
||||
},
|
||||
|
||||
async consumerGroupLag(clusterArn: string, groupId: string, topic?: string) {
|
||||
const admin = kafkaClientFactory(await brokers(clusterArn)).admin();
|
||||
await admin.connect();
|
||||
let completed = false;
|
||||
try {
|
||||
const offsets = await admin.fetchOffsets({ groupId, topics: topic ? [topic] : undefined });
|
||||
const lag = await Promise.all(offsets.map(async (topicOffsets) => {
|
||||
const endOffsets = await admin.fetchTopicOffsets(topicOffsets.topic);
|
||||
const endOffsetByPartition = new Map(endOffsets.map((offset) => [offset.partition, offset.offset]));
|
||||
const partitions = topicOffsets.partitions.map((offset) => ({
|
||||
partition: offset.partition,
|
||||
currentOffset: offset.offset,
|
||||
endOffset: endOffsetByPartition.get(offset.partition) ?? "0",
|
||||
lag: Math.max(0, Number(endOffsetByPartition.get(offset.partition) ?? 0) - Math.max(0, Number(offset.offset))),
|
||||
}));
|
||||
return {
|
||||
topic: topicOffsets.topic,
|
||||
partitions,
|
||||
totalLag: partitions.reduce((total, partition) => total + partition.lag, 0),
|
||||
};
|
||||
}));
|
||||
completed = true;
|
||||
return lag;
|
||||
} finally {
|
||||
try {
|
||||
await admin.disconnect();
|
||||
} catch (ignored) {
|
||||
// KafkaJS can close the coordinator socket after returning offsets.
|
||||
if (!completed) {
|
||||
throw ignored;
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export const mskService = createMskService();
|
||||
Reference in New Issue
Block a user