Fix backend Elasticsearch connectivity for local dev and app logs.
Auto port-forward using the registered cluster kubeconfig, default to loopback outside Kubernetes, route log APIs by app cluster, and document platform env settings. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
keyhan
@@ -87,6 +87,8 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
private readonly KIBANA_NAME = 'kibana';
|
||||
private portForwardChild: ChildProcess | null = null;
|
||||
private portForwardStartedByUs = false;
|
||||
private portForwardKubeconfigFile: string | null = null;
|
||||
private portForwardClusterId: string | null = null;
|
||||
private ensureInFlight: Promise<void> | null = null;
|
||||
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private healthCheckTimer: ReturnType<typeof setInterval> | null = null;
|
||||
@@ -133,6 +135,37 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
return host === '127.0.0.1' || host === 'localhost' || host === '::1';
|
||||
}
|
||||
|
||||
private isRunningInKubernetes(): boolean {
|
||||
return Boolean(process.env.KUBERNETES_SERVICE_HOST);
|
||||
}
|
||||
|
||||
/** Hostnames only resolvable from inside the workload cluster (not from a laptop). */
|
||||
private isClusterInternalHost(host: string): boolean {
|
||||
const h = host.toLowerCase();
|
||||
return (
|
||||
h.includes('svc.cluster.local') ||
|
||||
h.includes('.cluster.') ||
|
||||
h === 'elasticsearch' ||
|
||||
h === 'kibana'
|
||||
);
|
||||
}
|
||||
|
||||
private configuredElasticsearchHost(): string {
|
||||
return (
|
||||
this.configService.get<string>('elasticsearch.host') ||
|
||||
`${this.ES_NAME}.${this.ES_NAMESPACE}.svc.cluster.local`
|
||||
);
|
||||
}
|
||||
|
||||
/** HTTP target: loopback when we tunnel; cluster DNS when the API pod runs in-cluster. */
|
||||
private effectiveElasticsearchHost(): string {
|
||||
const configured = this.configuredElasticsearchHost();
|
||||
if (this.shouldAutoPortForward() && this.isClusterInternalHost(configured)) {
|
||||
return '127.0.0.1';
|
||||
}
|
||||
return configured;
|
||||
}
|
||||
|
||||
private shouldAutoPortForward(): boolean {
|
||||
if (this.configService.get<string>('elasticsearch.autoPortForward') === 'false') {
|
||||
return false;
|
||||
@@ -140,12 +173,15 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
if (process.env.ELASTICSEARCH_AUTO_PORT_FORWARD === 'false') {
|
||||
return false;
|
||||
}
|
||||
if (this.isRunningInKubernetes()) {
|
||||
return false;
|
||||
}
|
||||
const nodeEnv = process.env.NODE_ENV || 'development';
|
||||
if (nodeEnv === 'production') {
|
||||
return false;
|
||||
}
|
||||
const host = this.configService.get<string>('elasticsearch.host') || '';
|
||||
return this.isLoopbackHost(host);
|
||||
const host = this.configuredElasticsearchHost();
|
||||
return this.isLoopbackHost(host) || this.isClusterInternalHost(host);
|
||||
}
|
||||
|
||||
private stopDevPortForward(): void {
|
||||
@@ -154,9 +190,15 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
}
|
||||
const startedByUs = this.portForwardStartedByUs;
|
||||
const child = this.portForwardChild;
|
||||
const kubeconfigFile = this.portForwardKubeconfigFile;
|
||||
this.portForwardChild = null;
|
||||
this.portForwardStartedByUs = false;
|
||||
this.portForwardKubeconfigFile = null;
|
||||
this.portForwardClusterId = null;
|
||||
child.kill('SIGTERM');
|
||||
if (kubeconfigFile) {
|
||||
this.helmService.removeTempFile(kubeconfigFile);
|
||||
}
|
||||
if (startedByUs) {
|
||||
this.logger.log('Stopped Elasticsearch kubectl port-forward');
|
||||
}
|
||||
@@ -236,18 +278,36 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
return false;
|
||||
}
|
||||
|
||||
private startDevPortForward(localPort: number): void {
|
||||
if (this.portForwardChild) {
|
||||
private async startDevPortForward(localPort: number, clusterId?: string): Promise<void> {
|
||||
const targetClusterId = clusterId || null;
|
||||
if (
|
||||
this.portForwardChild &&
|
||||
this.portForwardClusterId === targetClusterId
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (this.portForwardChild) {
|
||||
this.stopDevPortForward();
|
||||
await new Promise((r) => setTimeout(r, 300));
|
||||
}
|
||||
|
||||
const { cluster } = await this.getK8sClients(clusterId);
|
||||
const kubeconfigFile = await this.helmService.createKubeconfigFile(cluster.kubeconfig);
|
||||
this.portForwardKubeconfigFile = kubeconfigFile;
|
||||
this.portForwardClusterId = targetClusterId;
|
||||
|
||||
const args = [
|
||||
'--kubeconfig',
|
||||
kubeconfigFile,
|
||||
'port-forward',
|
||||
'-n',
|
||||
this.ES_NAMESPACE,
|
||||
`svc/${this.ES_NAME}`,
|
||||
`${localPort}:9200`,
|
||||
];
|
||||
this.logger.log(`Starting kubectl ${args.join(' ')} (local log search)`);
|
||||
this.logger.log(
|
||||
`Starting kubectl port-forward to Elasticsearch on cluster "${cluster.name}" (local log search)`,
|
||||
);
|
||||
const child = spawn('kubectl', args, { stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
this.portForwardChild = child;
|
||||
this.portForwardStartedByUs = true;
|
||||
@@ -281,6 +341,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
*/
|
||||
private async ensureLocalElasticsearchAccess(options?: {
|
||||
waitForCluster?: boolean;
|
||||
clusterId?: string;
|
||||
}): Promise<boolean> {
|
||||
if (this.ensureInFlight) {
|
||||
await this.ensureInFlight;
|
||||
@@ -298,6 +359,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
|
||||
private async ensureLocalElasticsearchAccessImpl(options?: {
|
||||
waitForCluster?: boolean;
|
||||
clusterId?: string;
|
||||
}): Promise<void> {
|
||||
if (!this.shouldAutoPortForward()) {
|
||||
return;
|
||||
@@ -308,7 +370,8 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
return;
|
||||
}
|
||||
|
||||
let deployed = await this.isDeployed();
|
||||
const clusterId = options?.clusterId;
|
||||
let deployed = await this.isDeployed(clusterId);
|
||||
if (!deployed && options?.waitForCluster) {
|
||||
this.logger.log('Waiting for logging stack after cluster reconnect…');
|
||||
deployed = await this.waitForLoggingStack();
|
||||
@@ -325,7 +388,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
await new Promise((r) => setTimeout(r, 300));
|
||||
}
|
||||
|
||||
this.startDevPortForward(port);
|
||||
await this.startDevPortForward(port, clusterId);
|
||||
const ready = await this.waitForElasticsearch(90_000);
|
||||
if (ready) {
|
||||
this.reconnectAttempt = 0;
|
||||
@@ -341,16 +404,23 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
|
||||
private localElasticsearchHint(): string {
|
||||
const conn = this.getConnectionInfo();
|
||||
const configured = this.configuredElasticsearchHost();
|
||||
if (this.isLoopbackHost(conn.host)) {
|
||||
return (
|
||||
`Ensure port ${conn.port} is forwarded to the cluster (the API auto-starts kubectl port-forward in development). ` +
|
||||
`Manual: kubectl port-forward -n ${this.ES_NAMESPACE} svc/${this.ES_NAME} ${conn.port}:9200`
|
||||
);
|
||||
}
|
||||
if (conn.host.includes('svc.cluster.local') || conn.host.includes('.cluster.')) {
|
||||
if (this.isClusterInternalHost(configured)) {
|
||||
if (this.isRunningInKubernetes()) {
|
||||
return (
|
||||
'Ensure the logging stack is deployed on the same cluster as this API pod ' +
|
||||
`(Helm release ${LOGGING_HELM_RELEASE} in namespace ${LOGGING_HELM_NAMESPACE}).`
|
||||
);
|
||||
}
|
||||
return (
|
||||
'Run the API inside the cluster, or set ELASTICSEARCH_HOST=127.0.0.1 and keep port-forward running: ' +
|
||||
`kubectl port-forward -n ${this.ES_NAMESPACE} svc/${this.ES_NAME} ${conn.port}:9200`
|
||||
'Run the API inside the cluster, or restart the API locally so it can auto port-forward Elasticsearch ' +
|
||||
`(kubectl port-forward -n ${this.ES_NAMESPACE} svc/${this.ES_NAME} ${conn.port}:9200 using the cluster kubeconfig).`
|
||||
);
|
||||
}
|
||||
return `Ensure Elasticsearch is listening on ${conn.host}:${conn.port}.`;
|
||||
@@ -620,9 +690,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
*/
|
||||
getConnectionInfo(): { host: string; port: number; username: string; password: string } {
|
||||
return {
|
||||
host:
|
||||
this.configService.get<string>('elasticsearch.host') ||
|
||||
`${this.ES_NAME}.${this.ES_NAMESPACE}.svc.cluster.local`,
|
||||
host: this.effectiveElasticsearchHost(),
|
||||
port: this.configService.get<number>('elasticsearch.port') || 9200,
|
||||
username: 'elastic',
|
||||
password: this.ELASTIC_PASSWORD,
|
||||
@@ -791,6 +859,10 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
);
|
||||
}
|
||||
|
||||
if (this.shouldAutoPortForward()) {
|
||||
await this.ensureLocalElasticsearchAccess({ clusterId });
|
||||
}
|
||||
|
||||
const conn = this.getConnectionInfo();
|
||||
const url = `http://${conn.host}:${conn.port}${path}`;
|
||||
const auth = Buffer.from(`${conn.username}:${conn.password}`).toString('base64');
|
||||
@@ -800,7 +872,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
response = await this.elasticsearchFetch(url, auth, body);
|
||||
} catch (err: any) {
|
||||
if (this.shouldAutoPortForward()) {
|
||||
await this.ensureLocalElasticsearchAccess();
|
||||
await this.ensureLocalElasticsearchAccess({ clusterId });
|
||||
try {
|
||||
response = await this.elasticsearchFetch(url, auth, body);
|
||||
} catch {
|
||||
@@ -963,7 +1035,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
|
||||
}
|
||||
|
||||
if (this.shouldAutoPortForward() && !(await this.probeElasticsearch())) {
|
||||
void this.ensureLocalElasticsearchAccess();
|
||||
void this.ensureLocalElasticsearchAccess({ clusterId });
|
||||
}
|
||||
|
||||
if (await this.probeElasticsearch()) {
|
||||
|
||||
Reference in New Issue
Block a user