Downsides of Investigate a GO Memory Leak using PPROF

 

In this post we will review the limitations of using GO PPROF. A long time ago, I have listed Monitoring a GO application using pprof, which explains the basic steps to investigate a GO process. However when we encounter a memory leak, pprof might be misleading.

The Garbage Collector Issue

The first issue is that GO is a managed language: memory allocation and release is managed by the GO runtime, specifically by the garbage collector. So for any object we use in the process, GO keeps pointers and counters to manage it. This is reason that when we see in pprof heap analysis that the GO is using 1 GB of memory, the actual process memory might be double or even more.

The case might worsen in case we use many small objects. For example if we have a single string object of size 1M, the actual memory used will be very close to 1M, since the GO garbage collector should manage a single object. However if we have a JSON object or struct on 1M with many small elements and values, the GO garbage collector will keep pointers and counters for each of the elements, causing much higher memory usage. This is one of the reasons that in case possible, long term memory data is better to be stored as string or bytes and only converted to objects when required.

The Memory Allocation Issue

The second issue is what does PPROF heap supply? It provides a great graphical representation of memory allocations locations. This is different than memory references, and might be not useful. For example:

A GO process receives messages from NATS, deserialize them to structs using JSON unmarhsaling, process these structs and save some of the processed strings in baselines in the memory. Let's assume we have a bug in the baselines management that causes a leak. Where would the leak appear in pprof? In the memory allocation point, hence it will be shown in the deserializing code and not in the baseline management code. This is an important issue to understand.

Actually the GO garbage collector does have the referencing information, but pprof does not provide a way to get it, which is a shame. In case of need, we can create a full heap dump and analyze it using other external tools, which is not easy.

Final Note

We have presented two limitations of using GO pprof for analysis of a memory leak. While these are real issues that we face in a production application, pprof is still the best tool available. In future versions I hope GO maintainers would consider addressing these issues.

local-path-provisioner

 

The local-path-provisioner is a persistent volume provisioner dedicated for a kind based kubernetes cluster. Using it both simplifies helm charts and provides a better compatibility with EKS gp2 and gp3 storage clasess. In this post we will see how to use it.

Working Without a Provisioner

In case we don't have a provisioner we need to manually handle the persistent volume creation as part of the helm chart. This can be done using a new dedicated storage class for our kind development cluster. For example:

{{- if eq .Values.data.storageClass "development-storage" }}
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pv-1
  labels:
    type: local
spec:
  storageClassName: development-storage-class
  capacity:
    storage: 10G
  accessModes:
    - ReadWriteOnce
  hostPath:
    path: "/mnt/hostpath/data"
{{- end }}

Then we need to specify this storage class as part of the persistent volume claim:

  volumeClaimTemplates:
    - metadata:
        name: my-pvc
      spec:
        accessModes: [ "ReadWriteOnce" ]
        storageClassName: "development-storage-class"
        resources:
          requests:
            storage: 10G

Using The Local-Path-Provisioner

To use the local-path-provisioner we install it:

kubectl apply -f https://raw.githubusercontent.com/rancher/local-path-provisioner/master/deploy/local-path-storage.yaml

And then all we need it to specify local-path as the storage class, for example:

  volumeClaimTemplates:
    - metadata:
        name: my-pvc
      spec:
        accessModes: [ "ReadWriteOnce" ]
        storageClassName: "local-path"
        resources:
          requests:
            storage: 10G

We not only gain this simpler helm chart, but also permissions respectful provisioner. This simplifies life for pods that use modified security context such as grafana and clickhouse:

      securityContext:
        runAsUser: 50001
        runAsGroup: 50001
        fsGroup: 50001
        fsGroupChangePolicy: OnRootMismatch

Final Note

Actually, I guess I've got used to using the old method without the local-path-provisioner that I took it as is. Now that I know the local-path-provisioner, I think to myself, well why the complexity, I should have known it earlier.

PostgreSQL Implementation and Stub in GO

 

In this post we will display interface implementation and stub for PostgreSQL.

Interface

We include a query and a execute statement functions.

type PostgresqlApi interface {
    QueryRowsRw(
        destination any,
        sql string,
        args ...any,
    )

    ExecuteStatement(
        sql string,
        args ...any,
    )
}

Implementation

For PostgreSQL cluster we have 2 types of connections: First a connection to the read-write node, and second a connection to the replica of this node with is the read-only connection. To reduce the load, we can use the read-only connection if we can stand a short delay of non up to date data due to sync delay from the master node to the replica.

type PostgresqlImpl struct {
    rw      *pgxpool.Pool
    ro      *pgxpool.Pool
    context context.Context
}

func createConnection(
    connectionString string,
) *pgxpool.Pool {

    ctx, cancel := context.WithTimeout(
        context.Background(),
        Config.PostgresqlConnectionTimeout,
    )

    defer cancel()

    log.Info(
        "postgres connection string: %v",
        connectionString,
    )

    config, err := pgxpool.ParseConfig(connectionString)
    kiterr.RaiseIfError(err)

    config.ConnConfig.User = Config.PostgresqlUser
    config.ConnConfig.Password = Config.PostgresqlPassword

    log.V5(
        "postgres user: %v",
        config.ConnConfig.User,
    )

    log.V5(
        "postgres password: %v",
        config.ConnConfig.Password,
    )

    pool, err := pgxpool.NewWithConfig(ctx, config)

    if err != nil {

        kiterr.RaiseError(
            "connection to %v failed: %v",
            connectionString,
            err,
        )
    }

    return pool
}

func ProducePostgresqlImpl() *PostgresqlImpl {

    return &PostgresqlImpl{
        rw:      createConnection(Config.PostgresqlRwUrl),
        ro:      createConnection(Config.PostgresqlRoUrl),
        context: context.Background(),
    }
}

func (p *PostgresqlImpl) Close() {
    p.rw.Close()
    p.ro.Close()
}

For the query implementation we automatically convert the results to a struct with annotations. Example of usage is below.

type StepRow struct {
    SessionId   string `db:"session_id"`
    StepId      string `db:"step_id"`
    StepDetails string `db:"step_details"`
}

func LoadAllSteps(
    postgresApi postgres.PostgresqlApi,
) []*StepRow {

    var steps []*StepRow

    sql := `
SELECT
    session_id,
    step_id,
    step_details
FROM steps
`

    postgresApi.QueryRowsRw(&steps, sql)

    return steps
}

Stub

We running tests we do want to access an external process such as PostgreSQL. This is where SQLite power is shown. The stub uses in-memory SQLite database who supports the same mainstream SQL syntax as PostgreSQL. This is extremely powerful testing tool.

Here again we have the convert of the results into a struct.

import (
    "database/sql"
    "reflect"

    _ "github.com/mattn/go-sqlite3"
)

type PostgresqlStub struct {
    LogAction bool
    logger    *log.PrefixLogger
    db        *sql.DB
}

func ProducePostgresqlStub() *PostgresqlStub {

    db, err := sql.Open(
        "sqlite3",
        ":memory:",
    )

    kiterr.RaiseIfError(err)

    return &PostgresqlStub{
        LogAction: false,
        logger:    log.ProducePrefixLogger(
            "postgres stub ",
        ),
        db: db,
    }
}

Final Note
We have create a PostgreSQL wrapper and a stub. This hides the implementation details for the using code, and enables us to run both production code and tests code using the interface.

Deploying pgAdmin on Kubernetes

 

In this post we will review pgAdmin deployment on a kubernetes cluster. pgAdmin is a free open source tool enabling monitoring and development on PosgreSQL.

pgAdmin requires the components listed below.

The Service

apiVersion: v1
kind: Service
metadata:
  name: pgadmin-service
spec:
  selector:
    configid: pgadmin-container
  type: ClusterIP
  ports:
      - port: 80
        targetPort: 80
        name: tcp-api
        protocol: TCP

The Configmap

apiVersion: v1
kind: ConfigMap
metadata:
  name: pgadmin-configmap
data:
  servers.json: |
    {
      "Servers": {
        "1": {
          "Name": "Demo PostgreSQL",
          "Group": "Demo",
          "Host": "postgresql-rw",
          "Port": 5432,
          "MaintenanceDB": "demo",
          "Username": "admin",
          "Password": "admin",
          "SSLMode": "disable"
        }
      }
    }

This configmap enables us to add predefined PostgreSQL servers that the pgAdmin service can connect to.

The Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: pgadmin-deployment
spec:
  replicas: 1
  selector:
    matchLabels:
      configid: pgadmin-container

  template:
    metadata:
      labels:
        configid: pgadmin-container        
    spec:
      serviceAccountName: pgadmin-service-account
      containers:
        - name: pgadmin
          image: dpage/pgadmin4:9.15.0
          env:
            - name: PGADMIN_DEFAULT_EMAIL
              value: admin@pgadmin.com
            - name: PGADMIN_DEFAULT_PASSWORD
              value: admin
            - name: PGADMIN_LISTEN_ADDRESS
              value: "0.0.0.0"
          volumeMounts:
            - name: pgadmin-configmap
              mountPath: /pgadmin4/servers.json
              subPath: servers.json
      volumes:
        - name: pgadmin-configmap
          configMap:
            name: pgadmin-configmap

The user and password specified here are the login to the pgAdmin gui, and are not relevant to the PostgreSQL DB credentials.

Final Note

We have reviewed deployment of pgAdmin on a kubernetes cluster. In case we have a PostgreSQL in our solution, pgAdmin is a free and easy to use suite providing the tools to monitor and view the PostgreSQL database status and data. It is highly recommended to add it along the database.

Run Lambda from AWS Bedrock Agent

 

In this post we will review the step required to create an AWS bderock agent that run a lambda function. I am creating this post since AWS documentation is not sufficient to achieve this goal. I guess most software engineers today assume that you will use repeatadly use LLM such as ChatGPT to solve their no-so-good product and documentation issues.

Create an Agent

Start with agent creation, you can follow the steps in this post.

Create Action Group

Notice:
Make sure to select a model that you have permissions for. For example, in case you choose one of Anthropic models, you will probably fail with permissions since you did not purchased this model. And yes, this error will be just a general error forcing you to look for the actual failure reason yourself.

Next we need to create an action group: 

• Edit the agent, and under the action group section click add new action group
• Select "Define with API schemas" as the action group type
• Select "Quick create a new Lambda function" in the action group invocation
• Select "Define via in-line schema editor" in the action group schema

An example schema is the following:

openapi: 3.0.0

info:

  title: Person Info API

  version: 1.0.0

  description: API to get the person info

paths:

  /get_person_info:

    get:

      summary: Gets the person info

      description: Gets the person info

      operationId: get_person_info

      responses:

        '200':

          description: Gets the person info

          content:

            'application/json':

              schema:

                type: object

                properties:

                  info:

                    type: string

                    description: The person info

Create the Lambda

After clicking save, we can click on the button "View" next to the "Select Lambda function" combo. This opens a new tab where we can edit the lambda code. An example working lambda code is:

import logging

from typing import Dict, Any

from http import HTTPStatus

import json

logger = logging.getLogger()

logger.setLevel(logging.INFO)

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:

try:

action_group = event['actionGroup']

api_path = event['apiPath']

http_method = event['httpMethod']

message_version = event.get('messageVersion', '1.0')

# Your business logic output

result = {

"info": "this Person ate the entire cake"

}

response = {

"messageVersion": message_version,

"response": {

"actionGroup": action_group,

"apiPath": api_path,

"httpMethod": http_method,

"httpStatusCode": 200,

"responseBody": {

"application/json": {

"body": json.dumps(result)

}

}

}

}

logger.info("Response: %s", response)

return response

except Exception as e:

logger.error("Error: %s", str(e))

return {

"messageVersion": "1.0",

"response": {

"actionGroup": event.get("actionGroup", ""),

"apiPath": event.get("apiPath", ""),

"httpMethod": event.get("httpMethod", ""),

"httpStatusCode": 500,

"responseBody": {

"application/json": {

"body": json.dumps({"error": str(e)})

}

}

}

}

Notice:

Do not use the lambda code example that you get, it contains bugs.

Using the Lambda

Using the lambda is not straight forward. You need to perform additional 3 steps:

1. Deploy

After each update the the lambda code, click on Deploy.

Then, on the top of the screen, click on Actions, Publish new version.

2. Permissions

You need to add permissions to the agent role. Notice the permission includes the deployed version, so you need to rerurn this after each deployment. An example for permissions update is:

aws lambda add-permission   \ 

  --function-name get_person_info-3urcv   \

  --statement-id allow-bedrock-invoke-version-1 \

  --action lambda:InvokeFunction   \

  --principal bedrock.amazonaws.com   \

  --source-arn arn:aws:bedrock:us-east-1:123460611234:agent/*   \

  --qualifier 2   \

  --region us-east-1

3. Configure the agent

Edit the agent, then edit the action group, and replace the used lambda version under "Select Lambda function". Then save AND prepare the agent to use the updated configuration.

Final Note

Fun, it was not, but eventually it is working. We can now chat with the agent from the AWS console and using code like the one in the agent creation post.

Deploy a PostgreSQL Cluster on Kubernetes

 

In this post we will review the steps to deploy a PostgreSQL cluster on a kubernetes cluster. 

PostgreSQL is a veteran DBMS existing for ~30 years(!!). As such it is less a kubernetes player and does not automatically adjust the cluster like other technologies such as NATS and ClickHouse. Hence to maintain the cluster we require first to deploy an operator. There are several operators available, and in this post we use the CloudNativePG.

Cloud Native PG Operator

To deploy the CloudNativePG we can use the simple command:

kubectl apply -f https://raw.githubusercontent.com/cloudnative-pg/cloudnative-pg/release-1.29/releases/cnpg-1.29.0.yaml

However, in case of need to include this as part of an umbrella helm chart, we should download the file and split it to multiple templates. The CRDs should be included in a different folder, for example:

my-umbrella-helm/charts/cnpg/crds

The CRDs folder is a special helm folder enabling deployment of the CRDs before any additional templates are rendered.

Other template can be deployed in the standard templates folder, for example:

my-umbrella-helm/charts/cnpg/templates

We would usually update the templates in this folder to match our own helm deployment standards such as entities names, images location, labels, enable/disable flags. Notice that CloudNativePG requires the label:

app.kubernetes.io/name: cloudnative-pg

So we must include this label when customizing the labels.

The PostgreSQL cluster

Once the operator is up and running, all we need to do is to render a cluster CRD, for example:

apiVersion: postgresql.cnpg.io/v1
kind: Cluster
metadata:
  name: postgresql
spec:
  instances: 3

  imageName: my-repo/my-postgresql/dev:18.3
  imagePullPolicy: Always
  storage:
    size: 10Gi

  postgresql:
    parameters:
      max_connections: "200"

  bootstrap:
    initdb:
      database: my-db
      owner: admin
      secret:
        name: postgresql-secret

We usually update the image name to be downloaded from our own repo. Notice the CloudNativePG blocks usage of "latest" version for the PostgreSQL cluster, so even on the local kind deployment we must use a specific version.

In the cluster resource we specify several important settings: the default database name, the default owner role of the database, and a secret with the password for that user.

Final Note

We've reviewed steps to setup a PostgreSQL cluster using the CloudNativePG operator. While not as easy to setup as other DBMS, PostgreSQL is considered one of the best relational DBMS with a good support for updates.

MCP Client in GO and Stub

 

In the following post we present creation of MCP Client in GO including API, implementation, and stub. This method of providing APIs is critical for a successful testing which should not access any external entity. Note that in this case, we also provide a factory API, implementation, and stub further enabling the code to create MCP client even when it is running in the tests scope.

MCP Client

The client API:

type McpToolInfo struct {
  ToolName        string
  ToolDescription string
  InputSchema     string
  OutputSchema    string
}

type McpClientApi interface {
  ListTools() []*McpToolInfo
  CloseSession()
}

The implementation:

// static - to use connection pool for all servers
var staticWebClient = web.ProduceClientImpl(core.Config.McpConnectionTimeout)

type McpClientImpl struct {
  session *mcp.ClientSession
  context context.Context
}

func ProduceMcpClientImpl(
  mcpServerUrl string,
) *McpClientImpl {
  m := &McpClientImpl{
   context: context.Background(),
  }

  client := mcp.NewClient(
   &mcp.Implementation{
    Name:    "my-go-client",
    Version: "1.0.0",
   },
   nil,
  )

  transport := &mcp.StreamableClientTransport{
   Endpoint:   mcpServerUrl,
   HTTPClient: staticWebClient.UsedClient(),
  }

  session, err := client.Connect(m.context, transport, nil)
  kiterr.RaiseIfError(err)

  m.session = session
  return m
}

func (m *McpClientImpl) ListTools() []*McpToolInfo {
  parameters := mcp.ListToolsParams{}
  tools, err := m.session.ListTools(m.context, &parameters)
  kiterr.RaiseIfError(err)

  var result []*McpToolInfo
  for _, tool := range tools.Tools {
   info := McpToolInfo{
    ToolName:        tool.Name,
    ToolDescription: tool.Description,
    InputSchema:     fmt.Sprintf("%v", tool.InputSchema),
    OutputSchema:    fmt.Sprintf("%v", tool.OutputSchema),
   }
   result = append(result, &info)
  }

  return result
}

func (m *McpClientImpl) CloseSession() {
  err := m.session.Close()
  m.session = nil
  kiterr.RaiseIfError(err)
}

And the stub:

type McpClientStub struct {
}

func ProduceMcpClientStub(
  mcpServerUrl string,
) *McpClientStub {
  return &McpClientStub{}
}

func (m *McpClientStub) ListTools() []*McpToolInfo {
  return []*McpToolInfo{
   {
    ToolName:        "tool-1",
    ToolDescription: "the best tool",
    InputSchema:     "my-schema-input",
    OutputSchema:    "my-schema-output",
   },
  }
}

func (m *McpClientStub) CloseSession() {
}

MCP Client Factory

The factory API:

import "radware.com/mcpp/commons/mcp/client"

type McpClientFactoryApi interface {
  ProduceMcpClient(
   mcpServerUrl string,
  ) client.McpClientApi
}

The factory implementation:

type McpClientFactoryImpl struct {
}

func ProduceMcpClientFactoryImpl() *McpClientFactoryImpl {
  return &McpClientFactoryImpl{}
}

func (f *McpClientFactoryImpl) ProduceMcpClient(
  mcpServerUrl string,
) client.McpClientApi {
  return client.ProduceMcpClientImpl(mcpServerUrl)
}

And the stub:

type McpClientFactoryStub struct {
}

func ProduceMcpClientFactoryStub() *McpClientFactoryStub {
  return &McpClientFactoryStub{}
}

func (f *McpClientFactoryStub) ProduceMcpClient(
  mcpServerUrl string,
) client.McpClientApi {
  return client.ProduceMcpClientStub(mcpServerUrl)
}