Kubernetes Pod Can’t connect to DeepGram but works locally perfectly

[Hemendra05]

I'm banging my head against a wall with a network issue and hoping someone here can offer some insight. I've got an application that uses the Deepgram SDK for real-time transcription. It works flawlessly on my local machine, but when I deploy it as a pod in my Kubernetes cluster, it just… stops working.

Here's the breakdown:

Local: Everything's perfect. Audio streams in, transcripts come out.
Kubernetes: The pod receives the audio bytes input. No errors in my application logs related to authentication or API keys, after connection its also printing log for successful connection. Issue come when sending the bytes chunks to the DeepGram, it never happened.
I've tried a few things already:

Basic connectivity checks: Inside the pod, nslookup api.deepgram.com resolves correctly. I can even curl -v https://api.deepgram.com/, and get a 200 OK, which tells me basic HTTP traffic is going out.
Deepgram SDK/Socket API: Initially, I was using the Deepgram SDK. Thinking it might be an SDK issue, I tried working directly with WebSockets, but I get the same behavior: the connection gets closed immediately after the request.
Resource limits: I've checked CPU and memory limits on the pod; they're nowhere near being maxed out.
I'm clueless now about the issue. I'm attaching the code as well, if that makes sense.

Local Python Version: Python 3.11.4
Docker Python Image: python:3.11-slim
Kubernetes Version: 1.29.0
Kubernetes Ingress Annotations:

nginx.ingress.kubernetes.io/proxy-connect-timeout: "300"
nginx.ingress.kubernetes.io/proxy-read-timeout: "300"
nginx.ingress.kubernetes.io/proxy-send-timeout: "300"
nginx.ingress.kubernetes.io/proxy-websocket-timeout: "3600"
nginx.ingress.kubernetes.io/websocket-send-timeout: "300"
nginx.ingress.kubernetes.io/websocket-read-timeout: "300"

Code:

#sst_utils.py
import asyncio
import json
from typing import List
 
from app import logger
from deepgram import LiveTranscriptionEvents
 
from .connection_manager import STTConnectionManager, STTConnectionState
 
 
async def send_audio_to_deepgram(
    self,
    dg_connection,
    audio_queue: asyncio.Queue,
) -> None:
    """
    Continuously send audio chunks from the queue to Deepgram.
    If no audio arrives for 10 seconds, close the stream and stop.
    """
    try:
        while True:
            try:
                audio_bytes = await asyncio.wait_for(audio_queue.get(), timeout=10.0)
                logger.info(
                    "Sending audio chunk to Deepgram: %s bytes", len(audio_bytes)
                )
                await dg_connection.send(audio_bytes)
                logger.debug("Sent audio chunk to Deepgram.")
 
            except asyncio.TimeoutError:
                logger.info("No audio received for 10s, closing stream.")
                await dg_connection.finish()
                break
    except Exception as e:
        logger.error("Error sending audio to Deepgram: %s", e, exc_info=True)
        raise
    finally:
        logger.info("Sender task completed.")
 
 
async def receive_transcripts_from_deepgram(
    self,
    dg_connection,
    connection_manager: STTConnectionManager,
    connection_id: str,
) -> List[str]:
    """
    Receive transcripts from Deepgram, send them to the client, and accumulate final transcripts.
    """
    all_transcripts = []
 
    async def on_message(self, result, **kwargs):
        try:
            transcript = (
                result.channel.alternatives[0].transcript
                if result.channel.alternatives
                else ""
            )
            if transcript:
                logger.info("Final Transcript: %s", transcript)
                all_transcripts.append(transcript)
                response = {"transcript": transcript}
                connection_state = await connection_manager.get_connection_state(
                    connection_id
                )
                if connection_state:
                    await connection_manager.send_personal_message(
                        json.dumps(response), connection_state.websocket
                    )
        except Exception as e:
            logger.error("Error process message deepgram %s", e, exc_info=True)
 
    dg_connection.on(LiveTranscriptionEvents.Transcript, on_message)
    logger.info("Receiver task completed (no more messages from Deepgram).")
    return all_transcripts
 
 
async def close_deepgram_connection(self, connection_state: STTConnectionState):
    """Closes the Deepgram connection using the SDK method"""
    if connection_state.deepgram_ws:
        try:
            await connection_state.deepgram_ws.finish()
            logger.info(
                "Closed Deepgram connection for %s", connection_state.connection_id
            )
        except Exception as e:
            logger.error("Error closing Deepgram connection: %s", e, exc_info=True)
        finally:
            connection_state.deepgram_ws = None

#utils.py
import asyncio
 
from app import logger
from app.utils import (
    STTConnectionManager,
    STTConnectionState,
    close_deepgram_connection,
    receive_transcripts_from_deepgram,
    send_audio_to_deepgram,
)
from deepgram import DeepgramClient as DGClient  # TranscribeOptions,
from deepgram import LiveOptions
 
 
class DeepgramClient:
    """Manages communication with Deepgram's API using the SDK."""
 
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.dg_client = DGClient(api_key)
 
    async def process_audio_stream(
        self,
        connection_manager: STTConnectionManager,
        connection_id: str,
        connection_state: STTConnectionState,
    ) -> str:
        """Main function to handle STT streaming with Deepgram using SDK"""
        final_transcript = ""
        # if websocket connection is not created for Deepgram
        if not connection_state.deepgram_ws:
            try:
                logger.info("Creating Deepgram connection")
                options = LiveOptions(
                    model="nova-2",
                    language="en-IN",
                    encoding="linear16",
                    sample_rate=16000,
                )
                connection_state.deepgram_ws = self.dg_client.listen.asynclive.v("1")
                if await connection_state.deepgram_ws.start(options) is False:  # type: ignore
                    logger.error("Failed to connect to Deepgram")
                    return final_transcript
 
            except Exception as e:
                logger.error("Error creating Deepgram connection: %s", e, exc_info=True)
                return final_transcript
 
        try:
            sender_task = asyncio.create_task(
                send_audio_to_deepgram(
                    self, connection_state.deepgram_ws, connection_state.audio_queue
                )
            )
            receiver_task = asyncio.create_task(
                receive_transcripts_from_deepgram(
                    self,
                    connection_state.deepgram_ws,
                    connection_manager,
                    connection_id,
                )
            )
            done, pending = await asyncio.wait(
                [sender_task, receiver_task], return_when=asyncio.FIRST_EXCEPTION
            )
            for task in pending:
                task.cancel()
 
            for task in done:
                if task.exception():
                    raise task.exception()  # type: ignore
            all_transcripts = receiver_task.result()
            final_transcript = " ".join(all_transcripts)
        except Exception as e:
            logger.error(
                "Unexpected error with Deepgram connection: %s", e, exc_info=True
            )
        finally:
            await close_deepgram_connection(self, connection_state)
 
        return final_transcript

#conection_manager.py
 
import asyncio
import time
import uuid
from typing import Dict, Optional
 
import websockets
from app import logger
from fastapi import WebSocket
 
 
class STTConnectionState:
    """
    Data class to encapsulate connection state for each client
    """
 
    def __init__(
        self,
        websocket: WebSocket,
        audio_queue: asyncio.Queue,
        connection_id: Optional[str] = None,
    ):
        self.websocket: WebSocket = websocket
        self.audio_queue: asyncio.Queue = audio_queue
        self.connection_id = connection_id if connection_id else str(uuid.uuid4())
        self.is_connected = True
        self.last_activity = time.time()
        self.retries: int = 0
        self.max_retries: int = 3
        self.final_transcript: str = ""  # for storing final transcript for LLM
        self.deepgram_ws: Optional[websockets.WebSocketClientProtocol] = (  # type: ignore
            None  # store the deepgram websocket connection
        )
 
 
class STTConnectionManager:
    """Manages WebSocket connections and related state in memory."""
 
    def __init__(self):
        self.connection_states: Dict[str, STTConnectionState] = {}
 
    async def add_connection(self, websocket: WebSocket) -> STTConnectionState:
        """Adds a new connection and stores its state."""
        connection_state = STTConnectionState(websocket, asyncio.Queue())
        self.connection_states[connection_state.connection_id] = connection_state
        logger.info("Connection added: %s", connection_state.connection_id)
        return connection_state
 
    async def get_connection_state(
        self, connection_id: str
    ) -> Optional[STTConnectionState]:
        """Retrieves connection state from memory."""
        return self.connection_states.get(connection_id)
 
    async def update_connection_state(
        self, connection_state: STTConnectionState
    ) -> None:
        """Updates the connection state in memory."""
        self.connection_states[connection_state.connection_id] = connection_state
 
    async def remove_connection(self, connection_id: str) -> None:
        """Removes a connection from the manager."""
        if connection_id in self.connection_states:
            del self.connection_states[connection_id]
        logger.info("Connection removed: %s", connection_id)
 
    async def send_personal_message(self, message: str, websocket: WebSocket) -> None:
        """Sends a message to a specific WebSocket connection."""
        try:
            await websocket.send_text(message)
            logger.debug("Message sent to client: %s", message)
        except Exception as e:
            logger.error("Error sending message to client: %s", e, exc_info=True)
 
    async def handle_disconnect(self, connection_id: str) -> None:
        """Handles client disconnects."""
        connection_state = await self.get_connection_state(connection_id)
        if connection_state:
            connection_state.is_connected = False
            await self.update_connection_state(connection_state)
            logger.info("Client disconnected: %s", connection_id)
        else:
            logger.warning(
                "Client disconnect called for an unknown connection_id: %s",
                connection_id,
            )
 
    async def cleanup_inactive_connections(self, timeout_seconds: int = 30):
        """Clean up inactive connection states from memory"""
        now = time.time()
        inactive_connections = [
            connection_id
            for connection_id, state in self.connection_states.items()
            if now - state.last_activity > timeout_seconds and not state.is_connected
        ]
        for connection_id in inactive_connections:
            await self.remove_connection(connection_id)
 
    def get_active_connection_count(self) -> int:
        """Returns the count of the currently active connection in the memory"""
        return len(
            [state for state in self.connection_states.values() if state.is_connected]
        )

#main_endpoint.py
@app.websocket("/stt")
async def speech_to_text(websocket: WebSocket):
    """
    API endpoint for speech-to-text conversion.
    Accepts an audio input and returns the transcribed text.
    """
    await websocket.accept()
    connection_manager: STTConnectionManager = app.state.connection_manager
    deepgram_client: DeepgramClient = app.state.deepgram_client
 
    connection_state: Optional[STTConnectionState] = None
    deepgram_task: Optional[asyncio.Task] = None
 
    try:
        connection_state = await connection_manager.add_connection(websocket)
        connection_id = connection_state.connection_id
        logger.info("Client connected: %s", connection_id)
 
        deepgram_task = asyncio.create_task(
            deepgram_client.process_audio_stream(
                connection_manager, connection_id, connection_state
            )
        )
 
        while True:
            try:
                data = await websocket.receive_bytes()
                if data == b"CLOSE_CONNECTION":  # added for checking the end message
                    logger.info(
                        "Close connection request from client %s", connection_id
                    )
                    break
                connection_state.last_activity = time.time()
                await connection_manager.update_connection_state(connection_state)
                await connection_state.audio_queue.put(data)
 
            except WebSocketDisconnect:
                logger.info("Websocket disconnected: %s", connection_id)
                await connection_manager.handle_disconnect(connection_id)
                break
            except HTTPException as e:
                logger.error("Error receiving data: %s", e, exc_info=True)
                await connection_manager.handle_disconnect(connection_id)
                break
        if connection_state.is_connected:
            try:
                final_transcript = await deepgram_task
                connection_state.final_transcript = final_transcript
                await connection_manager.update_connection_state(connection_state)
                logger.info(
                    "Final transcript for %s: %s",
                    connection_id,
                    final_transcript,
                )
                await connection_manager.send_personal_message(
                    json.dumps({"transcript": final_transcript}), websocket
                )
            except HTTPException:
                raise
            except Exception as e:
                logger.error(
                    "Error processing deepgram stream %s: %s",
                    connection_id,
                    e,
                    exc_info=True,
                )
                raise
            finally:
                await close_deepgram_connection(
                    deepgram_client,
                    connection_state,
                )
                await connection_manager.remove_connection(connection_id)
 
    except HTTPException:
        raise
    except Exception as e:
        logger.error("Exception during connection lifecycle: %s", e, exc_info=True)
    finally:
        if deepgram_task and not deepgram_task.done():
            deepgram_task.cancel()
            try:
                await deepgram_task
            except asyncio.CancelledError:
                pass
 
        await connection_manager.cleanup_inactive_connections()
        logger.info(
            "Active connections in memory: %s",
            connection_manager.get_active_connection_count(),
        )

Thanks in advance for your help!

[deepgram-community[bot]]

Thanks for asking your question. Please be sure to reply with as much detail as possible so the community can assist you efficiently.
_{Consider joining our Discord community for more opportunity to engage with your fellow Deepgram users. You can earn points which can be redeemed for cool stuff by being active in our communities!}

[jkroll-deepgram]

Hi @Hemendra05, are you receiving an error message with a request ID from Deepgram? Or does the request fail before reaching Deepgram?

You mention that even curl -v https://api.deepgram.com/ succeeds from your pod. Can you try a pre-recorded curl request? See example commands in our docs. For instance:

curl \
  --request POST \
  --header 'Authorization: Token <YOUR_DEEPGRAM_API_KEY>' \
  --header 'Content-Type: application/json' \
  --data '{"url":"https://dpgr.am/spacewalk.wav"}' \
  --url 'https://api.deepgram.com/v1/listen?model=nova-2&smart_format=true'

Do you have any other networking restrictions in place on your pod?