DEV Community: Ernesto Ponce

Golang - AWS ECS lambda project

Ernesto Ponce — Sat, 20 Apr 2024 22:47:48 +0000

1. Introduction

Hi, in this tutorial we are going to create a project that allow us to upload a video using a endpoint HTTP and then convert it to another resolution format using another service , think of this as a process that a video platform like youtube o vimeo does when an user upload a video on their servers, in order to accomplish that we are going to use the following AWS services:

S3 Simple Storage Service:
We will use this service to store the videos in the original format and also in a compresses format, we are going to create two different buckets for each purpose.

Elastic Container Registry:
This service is for store our docker container images that will be use on ECS and lambda services.

ECS Elastic Container Service:
We will deploy our API http using ECS,this allows us to run our application in a scalable way without going into the complexity of managing the infrastructure manually, we will use containers to package our API service,

Lambda Serverless functions:
We will use this service to trigger a event/function when a video in the original format is uploaded to a bucket, this will obtain the video and convert it to another resolution ( using ffmpeg ) after save it on a different output bucket.

AFTER CREATED AND USE/TEST THE SERVICES, BE SURE TO DELETE THAT SERVICES FROM THE AWS DASHBOARD IN ORDER TO AVOID UNWANTED CHARGES.

Introduction
Create user on AWS IAM
Create S3 buckets
Create API service
Create ECR container image repository
Create ECS Elastic Container Service
Update API service
Create lambda container
Create lambda service
Conclusion

Diagram architecture

Requirements

AWS account - https://aws.amazon.com/
Docker - https://docs.docker.com/engine/install/
Go - https://go.dev/doc/install
AWS CLI - https://aws.amazon.com/cli/

We need to create an account in AWS in order to follow this tutorial,

Our services, API will use golang, so we need that on our machine.
We use docker to create our images, that images will be upload to ECR, and run on ECS and Lambda services.
Besides we will create most of the AWS resources using the web dashboard, we must install the AWS cli to upload the docker image.

Create user on AWS IAM

We need to create an user for this tutorial in AWS, this user will have an ACCESS KEY and a SECRET KEY that we will need later to connect to the AWS services that required our project, also we should limit the permissions and acces to this user to the specifics services that we need, this is very important for security reasons and for prevent any unwanted action on our account.

On the search box of AWS dashboard search for IAM and click on the first option that appears.

On IAM section go to.

Sidebar -> Access management -> Users -> Add user

On name put a name "aws-tutorial" or anything you want is not important at this point, after click on Next.

We now need to add the permissions that this user will have, we are going to add manually this settings, so check that option on the dasboard.

We need to add the following permissions policies:

AmazonS3FullAccess

This is required for our API services to upload files on S3 bucket.

AmazonEC2ContainerRegistryFullAccess

This is required to upload our docker images to the ECR service

Use the search box and select both policies, after that click on next.

Create acces key

After succesfuly create the user, enter on detail section of the user previously created.

Go to

Security credentials -> Create access key

On use cases, "SELECT OTHER"( AWS list other options and alternatives that we do not care at this moment ), click on Next, you can put a description if you
want, after that click on Create access key.

After that you see the access key and private key on the dashboard, copy that on some secure place of your own, this is because is the last time that AWS shows your private key, we need those values when we create our API and lambda services.

Create S3 buckets

Our project needs two buckets, one for the original video upload and another for the compress video files, for that go to the AWS dashboard and search for S3, click on the first option that appears.

Click on create bucket, a new form configuration appears, most of the options should work fine on default values, but we need to change the following options:

AWS Region: You can select any region you want, in this tutorial we are going to use "us-east-2 Oregon" region.

Bucket name: this must be a unique name, for this tutorial choose something that has the input name on it, for example "bucket-input", this semantic helps later when we create a trigger in the lambda server configuration.

We mantain the select option for "block all public access", we are going to access the bucket only by ours services, so we do not need to make it public.

After that click on create bucket.

For the second bucket who is the recipe for the compress videos, repeat the same steps as before, only remember to change the name for something like "bucket-output".

Create API service

For this tutorial we use golang to develop our API, we will create two endpoints, one that is in charge to received a file from a client and later save it on the bucket-input and another to show the API version text.

First a new project folder on your machine, name it ecs-tutorial, inside that folder run the following command.



go mod init ecs-tutorial

Install dependencies



go get github.com/aws/aws-sdk-go@v1.50.35
go get github.com/joho/godotenv@v1.5.1
go get github.com/go-chi/chi

aws-sdk: This is the AWS sdk that allow us to connect to different AWS services, in this case we are going to connect to S3 storage.

godotenv: This library allow us to use environment file (.env), we are going to use it for store our AWS credentials, version, port etc.

chi: This is a lightweight router HTTP, we are going to use it for create our API endpoints.

Start creating a .env file with this content.



AWS_S3_BUCKET=your-bucket-name
AWS_ACCESS_KEY=your-access-key
AWS_SECRET__KEY=your-secret-key
AWS_REGION=your-region
PORT=80

Replaces the values with your own, the bucket the we use in this service is the "input" bucket that we created before.

Create a main.go file with the following content.



package main

import (
    "log"
    "net/http"
    "os"

    "github.com/go-chi/chi"
    "github.com/go-chi/chi/middleware"
    "github.com/joho/godotenv"
)

func uploadFileHandler(w http.ResponseWriter, r *http.Request) {
    w.Write([]byte("upload"))
}

func main() {
    err := godotenv.Load()
    if err != nil {
        panic("Error loading .env file")
    }

    r := chi.NewRouter()
    r.Use(middleware.Logger)
    r.Get("/", func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("version 0.0.1"))
    })
    r.Post("/upload", uploadFileHandler)

    http.ListenAndServe(":"+os.Getenv("PORT"), r)
}

In the main function we first obtain the environment variables defined in the .env file, if that fails panic and stop the process,
after create a chi router object, we use the middleware logger for log server request info, that is helpful it on the terminal, lastly create two endpoints, one for the root path that show the version of the application, and another for the upload file,
at the moment funcion uploadFileHandler is return a hardcoded string "upload", later we will add the logic to save the file on the S3 bucket.

Create ECR container image repository

Create docker image

Before upload the image, we must create a docker image of our application, for that matter create a Dockerfile file on the ecs folder.

Dockerfile



# syntax=docker/dockerfile:1

# Build the application from source
FROM golang:1.22 AS build-stage

WORKDIR /app

COPY go.mod go.sum ./
RUN go mod download

COPY *.go ./

RUN CGO_ENABLED=0 GOOS=linux go build -o /ecs-api

FROM gcr.io/distroless/base-debian11 AS build-release-stage

WORKDIR /

COPY --from=build-stage /ecs-api /ecs-api
COPY .env ./

EXPOSE 80


ENTRYPOINT ["/ecs-api"]

We use the multi build docker buildkit feature in this definition,
first we create a image call build-stage, in this setup we use the official golang 1.21 version, next copy the go.mod and go.sum files and all the files .go that we have in our project, finally create the binary of our application using the go build command.

The second image is the release image, we use a distroless base image, this is a minimal image that only contains the necessary components to run our application, this has the advantege to be more secure and lightweight, next copy the binary create on the previous step, the .env file, and expose the port 80 ( this is the value the we have to define on the .env file ), the ENTRYPOINT value is the command that executes our binary.

Create image



docker build -t ecs-api .

Run the API server on background and expose service on port 8080



docker run -d -p 8080:80 ecs-api

Test service using curl



curl localhost:8080

That sould return the version of the API server.

Create ECR repository

On the AWS dashboard search for ECR and click on the first option that appears on the results,
select private access for the image and choose any name, for other options use the default values and click on Create Repository

We need to install the AWS cli on our machine, this is necessary to authenticate to our AWS account and upload the image on the registry.
So follow instructions on this page according to your OS.

https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html

Upload image to image registry

Go to the details of the repository that we created before, click on button "View push commands", this shows the commands that we need to execute on our machine to authenticate and upload our image in the registry.

You should see something like this.



aws ecr get-login-password --region us-west-2 | docker login --username AWS --password-stdin 99999.dkr.ecr.us-west-2.amazonaws.com

docker build -t ecs-api .

docker tag ecs-api yourawsuri:v0.0.1  

docker push yourawsuri:v0.0.1

After a successful push, we should see the image in the ECR aws dashboard, copy the value of the image URI , we need this when we create a ECS service, so copy that value on some place.

ECS - Create Elastic Container Service

ECS, or Amazon Elastic Container Service, is a fully managed container orchestration service provided by AWS. It allows developers to easily run, stop, and manage containers on a cluster. This service abstracts away the complexity of managing the underlying infrastructure, such as provisioning and scaling EC2 instances.

Go to the ECS section on AWS dashboard using the search box, click on Create Cluster, select any name you want, for example something like "ecs-tutorial",
in the Infrastructure section select the option AWS Fargate, this is a serveless feature that allow us not worry about the overhead of the maintance of the machines, mantain the other options with the default values and click on "Create".

Create task definition

We need to create a Task definition, this allow us to create a configuration for our Docker containers, such as CPU, memory, networking , image URI, etc.

Before that we must generate a "Task execution role", for that matter go to

IAM -> Access management -> Roles

Select AWS service as en entity type, in the use case select "Elasctic Container Service" -> Elastic Container Service Task click on Next.

In the permissions policies search box, type or copy "AmazonECSTaskExecutionRolePolicy", select that policy and click on Next.
This permission is required by our ECS cluster to create and have access to other AWS services.

Enter a role name and click on Create role, remember that value , because we will need that on next section.

On the ECS dashboard, click on Task Definitions sidebar, after go to Create new Task Definition,
This is a very long form, for most of the options we use the default value, but we need to change the following:

Task definition family: Put a name for the task definition, for example "aws-tutorial-task"

Infrastructure requirements: On Launch type select AWS Fargate Serverless,

Operating system/Architecture: Select Linux/x86_64, this is the architecture of our docker image.

Task size: Because this is a tutorial we are going to use the minimum values, so put 1GB for memory and 0.5 vCPU, change this values if you need more resources for your application.

Task execution role: Select the role that we created before, this is required for pulls them image for ECR, save logs to CloudWatch and other actions that required ECS to work properly.

Container details

Image URI: Select the URI of the image the we upload to ECR before.

Port mappings: Check this specific values.

port: 80

Protocol: TCP

App protocol: HTTP

Leaves the other containers configurations with the default values.

Storage, Monitoring and Tags are optional, so we can skip that sections and click on Create

Create service ECS

On ECS dashboard, select the cluster the we created before, go on service tabs and click on "Create" button.

Environment configuration:

Application type: Choose Service option, this is a best fit for our type of API long running application.

Task definition: On Family select the Task definition that we created in the previous step,
revisiion should selecte the last version of our task,

Service Name: Put a name for the service, for example "aws-tutorial-service"

Service type; Choose replica ,this allow us to run multiple copies of our application, and ECS will manage the load balancing and scaling for us.

Networking

On this section leave the VPC with the default values

Security group: Our services run on HTTP port 80, for that reasons we must create a new config with and onbound rule.

Make sure that public IP on turned on, this is because we are not going to use a Load Balancer in this tutorial, so we need to access our service using the public IP of the service.

Leave other options with default values and click on "Create"

Wait a moment and go to the cluster section, click on the Task tab service and go to the task attached to the service.

On the task detail, look up for the Public IP , you can test using curl



curl http://yourip/

That should return the api version response of our API service.

Update API service

In order to upload a file to a S3 bucket we need to update our handler uploadFileHandler with the following code.




import (
    "bytes"
    "context"
    "fmt"
    "io"
    "log"
    "net/http"
    "os"
    "time"

    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/awserr"
    "github.com/aws/aws-sdk-go/aws/credentials"
    "github.com/aws/aws-sdk-go/aws/request"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/s3"
    "github.com/go-chi/chi"
    "github.com/go-chi/chi/middleware"
    "github.com/joho/godotenv"
)

func uploadFileHandler(w http.ResponseWriter, r *http.Request) {
    // 10MB maximum file size
    err := r.ParseMultipartForm(10 << 20)
    if err != nil {
        http.Error(w, "Failed to parse multipart form", http.StatusBadRequest)
        return
    }

    file, fileHeader, err := r.FormFile("file")
    if err != nil {
        http.Error(w, "Failed to retrieve file from form data", http.StatusBadRequest)
        return
    }
    defer file.Close()

    fileBytes := bytes.Buffer{}
    _, err = io.Copy(&fileBytes, file)
    if err != nil {
        http.Error(w, "Failed to read file content", http.StatusInternalServerError)
        return
    }

    err = uploadToS3(fileHeader.Filename, bytes.NewReader(fileBytes.Bytes()))
    if err != nil {
        http.Error(w, "Failed to upload file to S3", http.StatusInternalServerError)
        return
    }

    w.WriteHeader(http.StatusOK)
    w.Write([]byte("File uploaded successfully"))
}

func uploadToS3(name string, fileContent io.ReadSeeker) error {
    bucket := os.Getenv("AWS_S3_BUCKET")
    key := os.Getenv("AWS_ACCESS_KEY")
    secret := os.Getenv("AWS_SECRET_KEY")
    timeout := 1000 * time.Second

    sess := session.Must(session.NewSession(&aws.Config{
        Region:      aws.String(os.Getenv("AWS_REGION")),
        Credentials: credentials.NewStaticCredentials(key, secret, ""),
    }))

    svc := s3.New(sess)

    ctx := context.Background()
    var cancelFn func()
    if timeout > 0 {
        ctx, cancelFn = context.WithTimeout(ctx, timeout)
    }

    if cancelFn != nil {
        defer cancelFn()
    }

    _, err := svc.PutObjectWithContext(ctx, &s3.PutObjectInput{
        Bucket:             aws.String(bucket),
        Body:               fileContent,
        Key:                aws.String(name),
        ContentDisposition: aws.String("attachment"),
    })
    if err != nil {
        if aerr, ok := err.(awserr.Error); ok && aerr.Code() == request.CanceledErrorCode {
            fmt.Fprintf(os.Stderr, "upload canceled due to timeout, %v\n", err)
            return err
        } else {
            fmt.Fprintf(os.Stderr, "failed to upload object, %v\n", err)
            return err
        }
    }

    return nil
}

On the handler code, we first set a file size limit of 10MB , after we check if the value "form-data" exists on the request and response a specific message if an error exists.

Next we copy the value of file content on a variable on type Buffer bytes, check for errors and if everything is ok call a function named
uploadToS3, this function recieve two parameters.

name: this is the name of the file that the client upload.

fileContent: this is a file object content that we are going to upload to S3, we need to convert the bytes to a Reader type in order to implement the io.ReadSeeker interface required by the AWS SDK.

We create a new AWS session using the AWS_ACCESS_KEY and the AWS_SECRET_KEY environment variables, next create a context with a timeout of 60 seconds that we use when we call the PutObjectWithContext method, this receives a bucket name, file content and the name of file, after that we check if the operation fails and returns and error value or nil if everything is ok.

Back in uploadFileHandler we check for the response for the uploadToS3 function, if an error exists we response a 500 status code and a message, otherwise we response a 200 status code and a message that the file was uploaded successfully.

Upload new image to ECR



docker build -t ecs-api .
docker tag ecs-api youtawsuri:v0.0.2
docker push youtawsuri:v0.0.2

Update ECS Task Definition

Back in the ECS dashboard, we should create a new task definition with the new image version, for that matter go to Task Definition -> Create New Revision, search the container config and change the value of the Image URI with the new image version.

After that, go to Cluster -> Service -> Update Service.

On this section, make sure that the revision value is the latest task definition we create before and click on Update,
after a few seconds we should see the new version deployed.

Check on the task detail the public IP and test using curl



curl -X POST -F "file=@./yourimage.png" http://yourip/upload

If this works correctly, you can see the file upload on the S3 input-bucket aws dasboard.

Create lambda container

We are going to create a container that will be running on a lambda function in AWS, this code will be called when we upload a video file on the input aws bucket.

Create a new folder call lambda on your machine, go to that folder and run this command.

Init project



go mod init lambda-tutorial

Install dependencies



go get github.com/aws/aws-lambda-go@v1.46.0
go get github.com/aws/aws-sdk-go@v1.50.35
go get github.com/joho/godotenv@v1.5.1

Create a main.go file with the following content.



package main

import (
    "context"

    "github.com/aws/aws-lambda-go/events"
    "github.com/aws/aws-lambda-go/lambda"
)

func handler(ctx context.Context, event events.APIGatewayProxyRequest) (events.APIGatewayProxyResponse, error) {
    response := events.APIGatewayProxyResponse{
        StatusCode: 200,
        Body:       "Hello lambda",
    }
    return response, nil
}

func main() {
    lambda.Start(handler)
}

On main function we call the Star method of the lambda sdk, this method requires a function argument call "handler" , it will be executed when the lambda function is invoked, this function uses a method of the AWS sdk called APIGatewayProxyResponse to create a response using a status code of 200 and a text response "Hello lambda".

Next, create a Dockerfile with the following content.



FROM golang:1.22 as build
WORKDIR /lambda
COPY go.mod go.sum ./
COPY main.go .
RUN go build -tags lambda.norpc -o main main.go

FROM public.ecr.aws/lambda/provided:al2023
COPY --from=build /lambda/main ./main
ENTRYPOINT [ "./main" ]

We use a multi-stage approach, first we define a build stage using a golang:1.22 image, copy the required files for the service go.mod, go.sum, main.go and build the binary of our lambda function, in the release stage we use an AWS official lambda image, this is required to work in the context of a lambda function service, copy the binary from the build stage en set the entrypoint command executing the binary.

Build container image



docker build -t lambda-tutorial .

Create container to test



docker run -d -p 9000:8080 \
--entrypoint /usr/local/bin/aws-lambda-rie \
lambda-tutorial ./main

Test using curl



curl -XPOST "http://localhost:9000/2015-03-31/functions/function/invocations" -d '{}'

The sould return this



{"statusCode":200,"headers":null,"multiValueHeaders":null,"body":"Hello lambda"}

Upload image to ECR

On the ECR AWS dashboard, create a new private repository called "lambda-tutorial" or something similar.

Build and upload the image to the registry, check view push commands on detail image.



docker build -t lambda-tutorial:v0.0.1 .
docker tag lambda-tutorial:v0.0.1 yourURI/lambda-tutorial:v0.0.1
docker push lambda-tutorial:v0.0.1 yourURI/lambda-tutorial:v0.0.1

Create lambda service

AWS Lambda is a serverless computing service that allows you to run code without provisioning or managing servers. It enables you to execute your code in response to various events, such as changes to data in S3 buckets, updates to DynamoDB tables, HTTP requests via API Gateway, or custom events generated by your applications.

Create policy

Go to IAM -> Access management -> Policies -> Create policy

On policiy editor section select JSON and paste this code



{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents",
                "logs:CreateLogGroup",
                "logs:CreateLogStream"
            ],
            "Resource": "arn:aws:logs:*:*:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject"
            ],
            "Resource": "arn:aws:s3:::*/*"
        }
    ]
}

This definition allow the lambda function to access to the Watch logs and S3 AWS services require by our lambda application.

Click on next, put a name like "lambda-tutorial-policy" and press on Create policy button.

Create role

Go to IAM -> Access management -> Roles -> Create role, select AWS service as entity type, in the use case select Lambda and click on next, on the permissions section search the policy that we created before "lambda-tutorial-policy" select it and click on next, in role detail select a name for the role, for example "lambda-tutorial-role" finally click on Create role.

Create lambda function

On the AWS dashboard go to the lambda section, make sure to select the same region that we used in the S3 bucked setup (in our case "us-west-2 - Oregon"), this is neccesary to make our trigger configuration works properly.

Select container image, put any function name, in the image URI you can type the value by hand or click in the "browse images" to select the version image ECR.

On Change default execution role, select Use an existing role and choose the execution role that we created before "lambda-tutorial-role".

Click on create, wait a moment until AWS finish the creation of the lambda ( you should see a message ), in order to check our container application , go to Test section and click on "Test" button, this should return the response "Hello lambda" that we define on the lambda function handler.

Add trigger s3

We need to configure our lambda to execute when a file is upload to a S3 bucket, for that reason go to the lambda detail and click on Add Trigger,

On Trigger configuration, select S3 as a trigger type, in the bucket select the S3 created on the first section of this tutorial, be sure to select the input bucket type, leave default values on the other options and click on Add.

Update dockerfile

We need to install the ffmepg in our image configuration , this is require to compress the video uploaded in the input-bucket, for that matter update Dockerfile with this content.



FROM golang:1.22 as build
WORKDIR /helloworld
COPY go.mod go.sum ./
COPY main.go .
RUN go build -tags lambda.norpc -o main main.go

FROM public.ecr.aws/lambda/provided:al2023
RUN dnf -y install wget && dnf -y install tar && dnf -y install xz && \ 
    cd /usr/local/bin && \
    mkdir ffmpeg && \ 
    cd ffmpeg && \
    wget https://www.johnvansickle.com/ffmpeg/old-releases/ffmpeg-4.2.1-amd64-static.tar.xz && \
    tar xvf ffmpeg-4.2.1-amd64-static.tar.xz && \
    ln -s /usr/local/bin/ffmpeg/ffmpeg-4.2.1-amd64-static/ffmpeg /usr/bin/ffmpeg
COPY .env .env
COPY --from=build /helloworld/main ./main
ENTRYPOINT [ "./main" ]

On the lambda image, we add a new RUN command, that will install some dependencies.

wget: required to download ffmpeg binaray file.

tar, xz: required to extract the file.

After installed those dependencies, we download the ffmpeg binary file using wget, extract the file and create a symbolic link to the /usr/bin/ffmpeg path, this is required in order to call the ffmpeg from our golang code.

Add a .env file with the definition of the required environment variables



AWS_S3_BUCKET=your-output-bucket
AWS_ACCESS_KEY=your-access-key
AWS_SECRET__KEY=your-secret-key
AWS_REGION=your-region

Update lambda/main.go



func convertFile(inputFile string, outputFile string) error {
    out, err := exec.Command("ffmpeg", "-i", inputFile, "-vf", "scale=640:480", outputFile).Output()

    if err != nil {
        fmt.Printf("ffmpeg command error:  %s", err)
        return err
    }

    fmt.Println("Command Successfully Executed")
    return nil
}

In this function we received as parameters the path string of the input and output video files that we use in a ffmepg call, this command compress the original video and save on the outputFile path, this is a very basic command that we use for this tutorial, ffmpeg has a lot of options and configurations for different resolutions and formats, you can check this link for more information.
Finally check for an error command and return that value of nil is everything is ok.

https://img.ly/blog/ultimate-guide-to-ffmpeg/#video-properties



func handlerTriggerS3Bucket(ctx context.Context, s3Event events.S3Event) error {
    err := godotenv.Load()
    if err != nil {
        log.Fatal("Error loading .env file")
    }

    bucketOutput := os.Getenv("AWS_S3_BUCKET")
    key := os.Getenv("AWS_ACCESS_KEY")
    secret := os.Getenv("AWS_SECRET_KEY")
    region := os.Getenv("AWS_REGION")
    timeout := 1000 * time.Second

    sess := session.Must(session.NewSession(&aws.Config{
        Region:      aws.String(region),
        Credentials: credentials.NewStaticCredentials(key, secret, ""),
    }))

    svc := s3.New(sess)
    var cancelFn func()
    if timeout > 0 {
        ctx, cancelFn = context.WithTimeout(ctx, timeout)
    }

    if cancelFn != nil {
        defer cancelFn()
    }

    for _, record := range s3Event.Records {
        bucket := record.S3.Bucket.Name
        key := record.S3.Object.URLDecodedKey

        obj, err := svc.GetObject(&s3.GetObjectInput{
            Bucket: aws.String(bucket),
            Key:    aws.String(key),
        })
        if err != nil {
            log.Printf("error getting head of object %s/%s: %s", bucket, key, err)
            return err
        }

        headOutput, err := svc.HeadObject(&s3.HeadObjectInput{
            Bucket: &bucket,
            Key:    &key,
        })
        if err != nil {
            log.Printf("error getting head of object %s/%s: %s", bucket, key, err)
            return err
        }
        log.Printf("successfully retrieved %s/%s of type %s", bucket, key, *headOutput.ContentType)

        var buf bytes.Buffer
        if _, err := io.Copy(&buf, obj.Body); err != nil {
            fmt.Fprintln(os.Stderr, "Error reading file:", err)
            return err
        }

        fmt.Println("File size (bytes):", buf.Len())

        inputFile := "/tmp/input-" + key
        outputFile := "/tmp/output-" + key
        err = os.WriteFile(inputFile, buf.Bytes(), 0644)
        if err != nil {
            fmt.Fprintln(os.Stderr, err)
            return err
        }

        err = convertFile(inputFile, outputFile)
        if err != nil {
            fmt.Fprintln(os.Stderr, err)
            return err
        }

        fileToUpload, err := os.ReadFile(outputFile)
        if err != nil {
            fmt.Fprintln(os.Stderr, err)
            return err
        }

        _, err = svc.PutObjectWithContext(ctx, &s3.PutObjectInput{
            Bucket:             aws.String(bucketOutput),
            Body:               bytes.NewReader(fileToUpload),
            Key:                aws.String(key),
            ContentDisposition: aws.String("attachment"),
        })
        if err != nil {
            if aerr, ok := err.(awserr.Error); ok && aerr.Code() == request.CanceledErrorCode {
                fmt.Fprintf(os.Stderr, "upload canceled due to timeout, %v\n", err)
                return err
            } else {
                fmt.Fprintf(os.Stderr, "failed to upload object, %v\n", err)
                return err
            }
        }
    }

    return nil
}


func main() {
    lambda.Start(handlerTriggerS3Bucket)
}

handlerTriggerS3Bucket is the function that will be called when a new file is upload in the input-bucket, first we created some helper variables that access to the values defined in the .env file, important thing, make sure that the value of AWS_S3_BUCKET is the correct output-bucket, to prevent a infinity loop lambda call in our AWS services.

After create a new session using the AWS sdk, this is the same code as the ECS container the we did in the previous section,
next we obtain the value of the record/file using the events.S3EventRecord type, call the GetObject method in order to retreived the object from S3.

Next call HeadObject method to get metadata of the object, this is only for debug purposes, we will see that value later in the CloudWatch logs.

We need to create a buffer variable to copy the value of the object S3 object, if no error happens on that call, we create and save a new file on the /tmp folder, next call the converFile function created previously that runs the ffmpeg command, if no error ocurred we read the output file that is created also in the /tmp folder and uploads the video compress in the output-bucket.

On main function we update the function trigger that is called on the lambda.Start method.

Build and upload the image to the registry, check the commands on the AWS dasboard ECR.



docker build -t lambda-tutorial:v0.0.2 .
docker tag lambda-tutorial:v0.0.1 yourURI/lambda-tutorial:v0.0.2
docker push yourURI/lambda-tutorial:v0.0.2

Update lambda image

Go to the AWS lambda and look up for the update image button

On Container Image setting -> Browse images, look for the lambda image and select the latest version, in this case v0.0.2, click on "Save", wait for a few seconds and we should see a success message.

Before test, we must to update the configuration of our lambda function, go to Configuration -> General configuration -> Edit.

Change the memory value to 256 MB and the timeout to 1 minute, this is required because the ffmpeg command is a complex operation and requires more resources than the default values, keep in mind that depending of the operation and size of the video file you should change this values accordingly.

Click on Save and wait a few seconds for the changes to take effect.

Test

We are going to use a video sample from this page

https://file-examples.com/index.php/sample-video-files/sample-mp4-files/

Select the 640x360 - size 3MB file , that should work fine with our configuration.

We can test using our API ECS service using curl



curl -X POST -F "file=@./yourvideo.mp4" http://yourip/upload

You can also use the S3 dasboard and upload directly in the input-bucket

S3 -> your-bucket -> Upload

After upload the video, the lambda function should be triggered, compress and upload the video in the output-bucket, you can check the CloudWatch logs for more information about the process.

Go to

Lambda -> your-lambda-function -> Monitor -> View logs in CloudWatch

Check logs

You also should see a new compress video file on your output-bucket, you can download and check the resolution and size of the file.

Conclusion

In this tutorial we learned how to use different AWS services to create a scalable and serverless application, we
create a API service using ECS, a lambda function that compress a video files, and a S3 bucket that store the original and compress video files.

Make sure to delete all the resources that we created in this tutorial.

Key value store in golang

Ernesto Ponce — Sat, 13 Jan 2024 19:16:46 +0000

KeyValue Database tutorial

In this tutorial we are going to create a simple key value database using go.
We can think in something like redis or etcd but with a very limited set of features and not ready for production.

But Why?

Because i think to create something from scratch is a great way to learn and understand how things works under the hood,
also if you know the basic on Golang and want to learn more about the language, create a real project like this is a great way to improve your skills, for that reason i am not go to deeply explaining sintax or basic concepts of the language.

How it works

The database will be a simple key value store, we will use a hash map to store the key that map to some data, we will use a file to persist data on disk.

We will use tests to check that everything works as expected after we add or modify the code, this is very important to prevent bugs and have more trust in our code.
also we will have a simple http api to interact with it, we will use the standard Go library to create the http server, although use HTTP is seems like an overhead, we are going to use it because is simpler and easy to implement instead of create a custom protocol like exits on redis, mysql, etcd or another database , and is pretty easy to test.

Example using curl

We send a json payload with a key, value data

curl -X POST -H "Content-Type: application/json" -d '{"key": "mykey", "value": "from curl"}' http://localhost:8000/set

Example get value by key

curl http://localhost:8000/get?key=foo

List of content

Create go project
Create engine package
Add tests
Persist data on disk
Set data on file
Get data from key
Compact data from file
Restore data from file on restore/start
Delete item
Create HTTP service
Update database files path

I am not expend much time explaining things abouts golang sintax so i assume that you have go installed, you have basic knowledge of the language and at least create some basic project, if you need a refresh take a look here https://gobyexample.com/.

Create go project

To start this project create a new folde called "keyvaluedb" go into that and create a new go module with the name keyvaluedb ( you can change this name for the name you want, if you do that remberber imports path where import local files)

mkdir keyvaluedb && cd keyvaluedb && go mod init keyvaluedb

Add a new file main.go and paste or write the following code to check at least that everything is working fine.

main.go

package main


import "fmt"

func main() {
    fmt.Println("Hello world")
}

Run the project

go run main.go

You should see the message Hello world print in the console, if not check your golang installation https://go.dev/doc/install.

Create engine package

We will create a new package called engine, this package will contain all the logic of the database, so for that create a new file engine.go and add the following code.

engine.go


package main

import (
    "errors"
    "sync"
)

type Engine struct {
    data map[string]string
}

func NewEngine() *Engine {
    return &Engine{
        data: make(map[string]string),
    }
}

func (e *Engine) Set(key, value string) error {
    e.data[key] = value
    return nil
}

func (e *Engine) Get(key string) (string, error) {
    value, ok := e.data[key]
    if !ok {
        return "", errors.New("key not found")
    }
    return value, nil
}

This code create a new struct called Engine, this will contain the data of the database, in this case we will use a simple map to store the data in memory, later we will add persistence on a file.

The NewEngine function create a new instance of the Engine struct, the important thing here is notice that we initialize the data map with make, this is a must because if we don't do this, the map will be nil and we will get a panic when we try to add a new key value pair.

The struct has two methods, Set and Get, Set will add a new key value pair to the map and Get will return the value of a key if exists,
if key not exists return an error.

In order to check , add this on main.go


func main() {
    e := NewEngine()
    e.Set("foo", "bar")
    value, err := e.Get("foo")
    if err != nil {
        panic(err)
    }
    fmt.Println(value)
}

Run the project using

go run .

This create a new instance on Engine, set a new key value pair and get the value of the key, if the key not exists, panic and print error, otherwise print the value on the console.

Add tests

Althoug we can use main.go for test the code, is a good idea to start using tests in order to check the code in a more consice, easy and stable way, is true that in this moment seems like an overkill, but because we will add more features to the code is a good idea to have automated tests from the beggining and check all with a single command.

Create a new file engine_test.go and add the following code

package main

import "testing"

func Test_SetGetKeyValue(t *testing.T) {
    e := NewEngine()
    e.Set("foo", "bar")
    value, err := e.Get("foo")
    if err != nil {
        t.Error(err)
    }
    if value != "bar" {
        t.Error("value should be bar")
    }

    _, err = e.Get("notfound")
    if err == nil {
        t.Error("should return error")
    }

}

This code create a new test function called Test_SetGetKeyValue, this function has a new instance of the Engine struct, set a new key value pair and get the value of the key, if the value is not the expected return an error,
we use the testing native native golang package to tests, we are not going to use any external libraries.

Run tests

go test

Persist data on disk

At the moment we only store key, value on memory, that works fine, but the problem is that if we for example restart the server or if a crash happened we lost all the data that we previously saved using Set method, in order to prevent that, the key/value data will be save on a file, this key value will be separate by a space and we differenciate items by a new line, for example

data.txt

foo bar
bar foo
user1 {"id": 1, "name": "ernesto"}

Set data on file

We are goint to update the Set method to save the data on file, the idea is to append the data at end of the file, using a concept called append only file, this is a common pattern used in some databases, where information can only be added or appended and not modified or deleted.
https://en.wikipedia.org/wiki/Append-only

engine.go

type Engine struct {
    data map[string]int64
    file *os.File
    mu   sync.Mutex
}

var keyValueSeparator = " "

func NewEngine() (*Engine, error) {
    file, err := os.OpenFile("data.txt", os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
    if err != nil {
        fmt.Println("Error opening file data:", err)
        return nil, err
    }

    return &Engine{
        data: make(map[string]int64),
        file: file,
        mu:   sync.Mutex{},
    }, nil
}

func (e *Engine) Set(key, value string) error {
    e.mu.Lock()
    defer e.mu.Unlock()

    offset, err := e.file.Seek(0, io.SeekEnd)
    if err != nil {
        fmt.Println("Error seeking file:", err)
        return err
    }

    _, err = e.file.WriteString(key + keyValueSeparator + value + "\n")
    if err != nil {
        fmt.Println("Error appending text:", err)
        return err
    }

    e.data[key] = offset
    return nil
}

In this code we update the Engine structs:
data: change the type of value map to a int64 instead of a string.
file: this property is a pointer to a os.File, this is used to read and write data from a file.
mu: this is for prevent concurrency problems when we write data to the file,

In the NewEngine function we open the file in read and write mode and configure to append data to the file when writing, if the file not exists create a new one, we also initialize the mutex property and initialize a map data structure.
In the Set function we use Lock in order to prevent problems when we write data, this is a must if we want to prevent conflicts when multiple clients try to write data to this file in the same moment, we use defer function to unlock the mutex when the Set method finish.
After we use the Seek function to move the cursor to the end of the file, this is because we need to append data to the file.
We use the WriteString function to write the key value pair to the file, we also add a new line at the end of the string, this is because we want to separate the key value pair by a new line.
Finally we now set the value of the map data with the value of the offset return by the Seek function, remember that with this new approach we need to search the data value on a file and not on memory.

Get data from key

Previusly we access to data directly for the map structure on memory, because in the previous step we start to save data in file , we need to change the Get method.

engine.go


func (e *Engine) Get(key string) (string, error) {
    e.mu.Lock()
    defer e.mu.Unlock()

    if _, ok := e.data[key]; !ok {
        return "", fmt.Errorf("key not found")
    }

    _, err := e.file.Seek(e.data[key]+int64(len(key))+1, 0)
    if err != nil {
        fmt.Println("Error seeking file:", err)
        return "", err
    }

    buffer := make([]byte, 1)
    var content []byte

    for {
        n, err := e.file.Read(buffer)
        if err != nil {
            fmt.Println("Error reading file:", err)
            break
        }

        if n == 0 {
            break
        }

        if buffer[0] == '\n' {
            break
        }

        content = append(content, buffer[0])
    }
    return string(content), nil
}

We validate if key exists on data map structure, if not exists, returns an error,
we use the Seek function to move the cursor to the offset of the key that is required by parameter on the method,
for example if we have the key-value "foo bar" saved on disk
we have to obtian the offset for key saved on the data map , plus the len of the key bar (4) plus (1) for the space separator, after that we have the cursor at the start of the value.

_, err := e.file.Seek(e.data[key]+int64(len(key))+1, 0)

after call this Seek method, the cursor of the file is at the start of the value of the key, with our example that should be on start of "bar"

Next we create a buffer of 1 byte, this is because we need to read the file byte by byte, we also create a content variable to save the value of the key in a slice of bytes, next we use a for loop to read the file, if we found a new line or reach the end of the file we break the loop, otherwise we append the byte to the content variable,
lastly the end of the method we transform the byte slice to a string and return the value.

Because we are now returning an error on the NewEngine function we should update our tests.
engine_test.go

func Test_SetGetKeyValue(t *testing.T) {
    e, _ := NewEngine()
    e.Set("foo", "bar")
    value, err := e.Get("foo")
    if err != nil {
        t.Error(err)
    }
    if value != "bar" {
        t.Error("value should be bar")
    }

    _, err = e.Get("notfound")
    if err == nil {
        t.Error("should return error")
    }

}

If now run tests, everything should be working fine.

go run test

Compact data from file

At this moment code is working well but we have a problem with the data, if we use the Set function multiple times with the same key, the value will be append to the file and multiple entries with the same key will be created, although the map will always return the last value of the key, we need to fix this problem and clean the file for old unused entries.

We need some refactor , first create a Compact function on engine.go


const Seconds = 5

func (e *Engine) CompactFile() {
    for {
        time.Sleep(time.Duration(Seconds) * time.Second)
        fmt.Println("Compacting file...")
        e.mu.Lock()

        _, m := e.GetMapFromFile()

        err = e.file.Truncate(0)
        if err != nil {
            fmt.Println(err)
            e.mu.Unlock()
            continue
        }

        for k, v := range m {
            e.setRaw(k, v)
        }

        e.file.Seek(0, 0)
        e.mu.Unlock()
    }
}

This method will be run as a background job using a goroutine, like this

go e.CompactFile()

We get the map of the file using the GetMapFromFile method, next we truncate the original file,
after that we loop over the map data and use the setRaw function to write the data to the file, finally we move the cursor to the start of the file and unlock the mutex for future uses, if we have any error we must unlock the mutex and continue with the loop.
that works because the map data only have the latests and valid key values pairs.

Add function GetMapFromFile in engine.go


type Item struct {
    Key    string
    Value  string
    Offset int64
}

func (c *Engine) GetMapFromFile() ([]Item, map[string]string) {
    m := make(map[string]string)
    i := []Item{}

    _, err := c.file.Seek(0, 0)
    if err != nil {
        fmt.Println(err)
        return i, m
    }

    var totalBytesRead int64
    scanner := bufio.NewScanner(c.file)

    for scanner.Scan() {
        line := scanner.Text()
        offset := totalBytesRead
        parts := strings.Split(line, keyValueSeparator)
        if len(parts) >= 2 {
            m[parts[0]] = parts[1]
            i = append(i, Item{
                Key:   parts[0],
                Value: parts[1],
                Offset: offset,
            })
        }
    }

    return i, m
}

We created a Item struct for the key, value data, on the GetMapFromFile method we create a map and a slice of Item, after that we move the cursor to the start of the file and we use a scanner to read the file line by line, we split the line by the space separator and save the key, value on the map and the slice of Item, finally we return the slice of Item and the map for later uses.

We need to set the offset value on the Item struct, this is the value of the key map struct the we use to get items from database, we created a totalBytesRead variable that will be used to calculate the offset of the key, to obtain that we sum the len of the line plus one for the new line character, on every iteration of the loop we set the offset value and create a Item struct.

update methods in engine.go

func (e *Engine) Set(key string, value string) error {
    e.mu.Lock()
    defer e.mu.Unlock()

    if strings.Contains(key, " ") {
        return fmt.Errorf("key cannot contain spaces")
    }

    return e.setRaw(key, value)
}

func (e *Engine) setRaw(key string, value string) error {
    offset, err := e.saveToFile(key, value)
    if err != nil {
        return err
    }

    e.setKey(key, offset)
    return nil
}

func (e *Engine) setKey(key string, value int64) {
    e.data[key] = value
}

func (c *Engine) saveToFile(key string, value string) (int64, error) {
    offset, err := c.file.Seek(0, io.SeekEnd)
    if err != nil {
        fmt.Println("Error seeking file:", err)
        return 0, err
    }

    _, err = c.file.WriteString(key + keyValueSeparator + value + "\n")
    if err != nil {
        fmt.Println("Error appending text:", err)
        return 0, err
    }

    return offset, nil
}

On set function we check if the key contains spaces, we must do that because we use a blank space as a separator of key/value on the file, next we call a method call setRaw, that uses this other methods.

saveToFile: this function save data in file and return the offset of the key on the file.
setKey: this function save the key and value offset on the map.

Add this test on engine_test.go

func (c *Engine) GetFileContent(f *os.File) []string {
    c.mu.Lock()
    defer c.mu.Unlock()

    _, err := f.Seek(0, 0)
    if err != nil {
        fmt.Println(err)
        return []string{}
    }

    scanner := bufio.NewScanner(f)

    var content []string
    for scanner.Scan() {
        line := scanner.Text()
        content = append(content, line)
    }

    return content
}

We also add a method call GetFileContent, this is a helper function that we use to get the content of a file, we use this function on the tests.

To check all this changes create a new test

engine_test.go

func TestEngine_Compact(t *testing.T) {
    os.Remove("data.txt")
    v1 := "latestvalue1"
    v2 := "latestvalue2"
    e, _ := NewEngine()
    e.Set("key1", "value1")
    e.Set("key2", "value2")
    e.Set("key1", v1)
    e.Set("key2", v2)
    e.Set("key3", "value3")

    go e.CompactFile()

    time.Sleep((Seconds + 3) * time.Second)
    if len(e.GetFileContent(e.file)) != 3 {
        t.Errorf("Expected %d, but got %d", 3, len(e.GetFileContent(e.file)))
    }

}

First we remove the data.txt file, later we will fix that and uses another specific file for testing,
After we set some repeated keys with diferent values, this will create a file with this data

key1 value1
key2 value2
key1 latestvalue1
key2 latestvalue2
key3 value3

we see that we have some keys repeated, in order to clean that we call the CompactFile function using a goroutine, this runs every 5 seconds.
we wait for that to run and check count of lines/data on the data.txt file,
that should be 3, and have this data.

key1 latestvalue1
key2 latestvalue2
key3 value3

Run test

go test

Restore data from file on restore/start

At this moment the project works fine, we can set, get value and run tests, but we have something to pay attention, if we restart or the server/process crash, we lost all the data on memory, especifically key value saved on map structure.
we need to restore the data that is saved on the file on the map memory in order to Set method workd normally.

add this function on engine.go

func (e *Engine) Restore() {
    e.mu.Lock()
    defer e.mu.Unlock()

    items, _ := e.GetMapFromFile()

    for _, v := range items {
        e.setKey(v.Key, v.Offset)
    }
}

func (c *Engine) Close() {
    c.file.Close()
}

This method read data from database file and get a map calling the method GetMapFromFile, after that we loop over the map and save the key, value on the map using the setKey method that we create before, like in all methods that write or read a file we use a mutex to prevent concurrency problems.

The Close method close the file when program finish, this is necessary to prevent memory leaks.

Add this tests on engine_test.go.

func TestEngine_Restore(t *testing.T) {
    os.Remove("data.txt")
    e, _ := NewEngine()

    e.Set("key1_restore", "value1")
    e.Set("key2_restore", "value2")

    e.Close()

    e, _ = NewEngine()
    e.Restore()
    k, _ := e.Get("key1_restore")

    if k != "value1" {
        t.Errorf("Expected %s, but got %s", "value1", k)
    }
}

in this test we remove the data.txt file, after that we set some values, close the file and create a new Engine object (this simulates the creation of a new instance after a crash), next we call the Restore function and get the value of a key, if the value is not the expected return an error.

Delete item

Next feature will be an option to delete keys, the API usage is this

e := NewEngine()
e.Set("foo", "bar")
e.Delete("foo")

We need to create another file in order to track all keys that need to be remove, this job will be execute in an asyncronous way, we will use a goroutine to do that.

We need to make updates on engine.go

type Engine struct {
    data       map[string]int64
    file       *os.File
    fileDelete *os.File
    mu         sync.Mutex
    muDelete   sync.Mutex
}

var keyValueSeparator = " "

func NewEngine() (*Engine, error) {
    file, err := os.OpenFile("data.txt", os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
    if err != nil {
        fmt.Println("Error opening file data:", err)
        return nil, err
    }

    fileDelete, err := os.OpenFile("delete.txt", os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
    if err != nil {
        fmt.Println("Error opening file delete:", err)
        return nil, err
    }

    return &Engine{
        data:       make(map[string]int64),
        file:       file,
        fileDelete: fileDelete,
        mu:         sync.Mutex{},
    }, nil
}

First we add a new property on Engine struct called fileDelete, this is a pointer to a os.File, this will be used to save the keys that need to be deleted, we also add a new mutex called muDelete.

Add this new method on engine.go

func (e *Engine) Delete(key string) error {
    e.muDelete.Lock()
    defer e.muDelete.Unlock()
    _, err := e.fileDelete.Seek(0, io.SeekEnd)
    if err != nil {
        fmt.Println("Error seeking file:", err)
        return err
    }

    _, err = e.fileDelete.WriteString(key + "\n")
    if err != nil {
        fmt.Println("Error writing to file:", err)
        return err
    }

    e.mu.Lock()
    defer e.mu.Unlock()
    delete(e.data, key)

    return nil
}

In this function we fist lock our delete file mutex, after we move the cursor to the end of the file and write the key that we obtain from the parameter, is the same approach that we use when we save key/value data on the other file.
If everything works fine we delete the key from the map, here we must use the other mutext because we are making changes on the map structure of data,

Add test for this


func TestEngine_DeleteKey(t *testing.T) {
    os.Remove("data.txt")
    os.Remove("remove.txt")
    e, _ := NewEngine()

    e.Set("key1_delete", "value1")
    e.Set("key2_delete", "value2")

    err := e.Delete("key1_delete")
    if err != nil {
        panic(err)
    }

    k, _ := e.Get("key1_delete")

    if k != "" {
        t.Errorf("Expected %s, but got %s", "", k)
    }

    if len(e.GetFileContent(e.fileDelete)) != 1 {
        t.Errorf("Expected %d, but got %d", 1, len(e.GetFileContent(e.file)))
    }
}

We add remove call for the delete file in order to have a clean state on tests,
after that we set some values, delete one of them and check that the key not exists on the map, lastly check that the delete file has only one entry.

At the moment we delete the item for the map and add a entry on the delete.txt file, but we are not doing anyting with the data.txt file, for that matter we need to create a couple of function that resolves that.


const secondsDelete = 5
func (e *Engine) DeleteFromFile() {
    for {
        time.Sleep(secondsDelete * time.Second)
        fmt.Println("Deleting from file...")
        e.muDelete.Lock()

        _, err := e.fileDelete.Seek(0, 0)
        if err != nil {
            fmt.Println(err)
            e.muDelete.Unlock()
            continue
        }

        scanner := bufio.NewScanner(e.fileDelete)

        content := []string{}
        for scanner.Scan() {
            line := scanner.Text()
            if line != "" {
                content = append(content, line)
            }
        }

        err = e.deleteKeyFromFile(content)
        if err != nil {
            fmt.Println(err)
            e.muDelete.Unlock()
            continue
        }

        err = e.fileDelete.Truncate(0)
        if err != nil {
            fmt.Println(err)
            e.muDelete.Unlock()
            continue
        }

        e.muDelete.Unlock()
    }
}

This method runs on background and will be called using a goroutine,
we create a variable secondsDelete that will be used to configure the time that the method will wait to run again,
after we create a loop that uses the mutex created for the delete file,
read the file line by line and save the content on a slice of string, after we call a function called deleteKeyFromFile ( next to analyse ) that will make the changes in order to delete the keys found on the data.txt
finally we truncate the delete file and unlock the mutex.

func (c *Engine) deleteKeyFromFile(keys []string) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    _, err := c.file.Seek(0, 0)
    if err != nil {
        fmt.Println(err)
        return err
    }

    var bs []byte
    buf := bytes.NewBuffer(bs)

    scanner := bufio.NewScanner(c.file)
    for scanner.Scan() {
        l := scanner.Text()

        parts := strings.Split(l, keyValueSeparator)
        if len(parts) >= 2 {
            found := false
            for _, k := range keys {
                if parts[0] == k {
                    found = true
                    break
                }
            }

            if !found {
                buf.WriteString(l + "\n")
            }
        }
    }

    _, err = c.file.Seek(0, 0)
    if err != nil {
        fmt.Println(err)
        return err
    }

    err = c.file.Truncate(0)
    if err != nil {
        fmt.Println(err)
        return err
    }


    _, err = buf.WriteTo(c.file)
    if err != nil {
        fmt.Println(err)
        return err
    }

    return nil
}

This method recieved a keys parameters, this is a slice of string that contains the keys that need to be deleted,
we lock the mutex of the data.txt file, after that we move the cursor to the start of this file,
we also create a buffer variable of bytes, this will be used to get the items that not need to be deleted,
we check that looping on the keys parameter and validating key of the data.txt,
if variable found is false we use the writeString function to write the line to the buffer, otherwise we continue with the loop.

After that we have the variable buffer with the data that is not mark to be deleted, we move the cursor to the start of the file, truncate that and finally copy the content of the buffer on the data.txt.

add this test

func TestEngine_DeleteKeyFromFile(t *testing.T) {
    os.Remove("data.txt")
    os.Remove("delete.txt")
    e, _ := NewEngine()

    e.Set("key1_delete", "value1")
    e.Set("key2_delete", "value2")
    e.Set("key3_delete", "value3")

    e.deleteKeyFromFile([]string{"key2_delete", "key3_delete"})

    if len(e.GetFileContent(e.file)) != 1 {
        t.Errorf("Expected %d, but got %d", 1, len(e.GetFileContent(e.file)))
    }
}

This test create some keys and call the deleteKeyFromFile function with two keys that need to be deleted, after that we check that the delete.txt file has only one entry.

Create HTTP service

At the moment we are testing the code of engine.go using tests, that is great because you can check all the feautures d the project with a single command and gain more confidence in your code when you need to make changes or updates, but we do not have any service or way to interact with our database from external clients, in order to change that, we are going to create a simple http server, we will expose three endpoint to create, get and delete keys on the database.

On main.go add this code.

package main

import (
    "fmt"
    "net/http"
)

func handlerSet(w http.ResponseWriter, r *http.Request)    {}
func handlerGet(w http.ResponseWriter, r *http.Request)    {}
func handlerDelete(w http.ResponseWriter, r *http.Request) {}

var e *Engine

func main() {
    var err error
    e, err = NewEngine()
    if err != nil {
        panic(err)
    }
    defer e.Close()
    e.Restore()

    go e.CompactFile()
    go e.DeleteFromFile()

    http.HandleFunc("/set", handlerSet)
    http.HandleFunc("/get", handlerGet)
    http.HandleFunc("/delete", handlerDelete)

    address := ":8080"

    fmt.Printf("Server is listening on http://localhost%s\n", address)
    err = http.ListenAndServe(address, nil)
    if err != nil {
        fmt.Println("Error:", err)
    }
}

In this code we create three functions that will be used as handlers for the http server, at the moment are empty, we will change that in the future.
On main function we first create a instance of the Engine DB ( we use a global e variable to get a more easy access on the handlers) and panic if there is an error,
after we call the restore function, we have to run this with the service start in order to recover for crashes or error, this get the data from the file and save it on the map data structure,

next we call the CompactFile, and DeleteFromFile on background using a goroutine,
CompactFill will remove duplicate values on the file database,
DeleteFromFIle will remove keys-value from the file database,
Lastly we create the routes on the http server and start on port 8080.

Update handlers

We must update the handlers create previously on main.go, the basic idea is to use the methods of the engine file and respond a JSON in an endpoint http.

main.go


type RequestPayload struct {
    Key   string `json:"key"`
    Value string `json:"value"`
}

type ResponseJson struct {
    Status  string `json:"key"`
    Message string `json:"value"`
}

func handlerSet(w http.ResponseWriter, r *http.Request) {
    if r.Method == "POST" {
        body, err := io.ReadAll(r.Body)
        if err != nil {
            http.Error(w, "Error reading request body", http.StatusBadRequest)
            return
        }

        var rp RequestPayload

        err = json.Unmarshal(body, &rp)
        if err != nil {
            http.Error(w, "Error decoding JSON", http.StatusBadRequest)
            return
        }

        err = e.Set(rp.Key, rp.Value)
        if err != nil {
            responseJSON(w, ResponseJson{
                Status:  "error",
                Message: err.Error(),
            }, http.StatusInternalServerError)
            return
        }

        responseJSON(w, ResponseJson{
            Status:  "success",
            Message: "Key value pair saved successfully.",
        }, http.StatusOK)
    } else {
        fmt.Println("Invalid request method.")
        fmt.Fprintf(w, "Invalid request method.")
    }
}

func responseJSON(w http.ResponseWriter, data interface{}, status int) {
    d, err := json.Marshal(data)
    w.Header().Set("Content-Type", "application/json")

    if err != nil {
        w.WriteHeader(http.StatusInternalServerError)
        w.Write([]byte("Internal server errror"))
        return
    }

    w.WriteHeader(status)
    w.Write(d)
}

We created a few new structs, RequestPayload is needed to decode the JSON payload sent by the client
and Response is used to send a JSON response to the client.
In the function handlerSet we first check the method of the request, if is not a POST request we return an error,
We read the Body of the request and decode the JSON payload to the RequestPayload struct and check for errors,
after that we call the Set function of the engine file, we now return a response using a function called responseJSON the we will check next.

ResponseJSON is a helper function that we used to prevent repeat code multiple times, this function receive a interface (basically any struct is valid), if something goes wrong we return a 500 error,
otherwise we set a status code with the value received from the argument, set the content type to application/json and write the JSON response to the client.

We can test this using curl

go run .

curl -X POST -H "Content-Type: application/json" -d '{"key": "mykey", "value": "myvalue"}' http://localhost:8080/set

This should return a json success message and save data on the data.txt database file

Get handler


func handlerGet(w http.ResponseWriter, r *http.Request) {
    if r.Method == "GET" {
        key := r.URL.Query().Get("key")
        value, err := e.Get(key)
        if err != nil {
            responseJSON(w, ResponseJson{
                Status:  "error",
                Message: err.Error(),
            }, http.StatusNotFound)
            return
        }

        responseJSON(w, RequestPayload{
            Key:   key,
            Value: value,
        }, http.StatusOK)
    } else {
        fmt.Println("Invalid request method.")
        fmt.Fprintf(w, "Invalid request method.")
    }
}

In this function we first check the method of the request, if is not a GET request we return an error like in the set function,
next we get the key from the query params and uses the Get function of the engine instance, if something goes wrong we return a 404 error, otherwise we return a JSON response with the key and value.

Test with curl

go run .
curl http://localhost:8080/get?key=mykey

Delete handler

func handlerDelete(w http.ResponseWriter, r *http.Request) {
    if r.Method == "DELETE" {
        key := r.URL.Query().Get("key")
        err := e.Delete(key)
        if err != nil {
            responseJSON(w, ResponseJson{
                Status:  "error",
                Message: err.Error(),
            }, http.StatusInternalServerError)
            return
        }

        responseJSON(w, ResponseJson{
            Status:  "success",
            Message: "Key deleted successfully.",
        }, http.StatusOK)
    } else {
        fmt.Println("Invalid request method.")
        fmt.Fprintf(w, "Invalid request method.")
    }
}

This function is very similar to the Get function, the only difference is that we use the Delete function of the engine instance instead and change the message response

test in curl

go run .
curl -X DELETE http://localhost:8080/delete?key=mykey

Update database files path

Currently we have the project running using tests and also exposing a http server, this works we have some problems with this approach.
We are using the same files for tests and for the server, so when we run a test we are modifying and deleting the data created via API endpoints.
The file is save relative to the current path in which the project is running, this is not a good thing, because if we start the service from another path we will create a new Database file.

To fix that we we use this approach,
in tests, we must define the name of the files data and delete and for the server and not tests instances we must use the path of the current user running the project, and save this on a folder called keyvaluedb the lives in .config home user folder.

The .config values is uses for multiple applications to save data, for example discord, chrome, VirtualBox, etc save files on this place.

Check in your machine with this command

ls ~/.config

update engine.go

type Config struct {
    FileData   string
    FileRemove string
}

var keyValueSeparator = " "

func NewEngine(cfg Config) (*Engine, error) {
    if cfg.FileData == "" && cfg.FileRemove == "" {
        configFolderPath, err := getConfigFolder()
        if err != nil {
            fmt.Println(err)
            return nil, err
        }

        if _, err := os.Stat(configFolderPath); os.IsNotExist(err) {
            err := os.Mkdir(configFolderPath, 0700)
            if err != nil {
                fmt.Println(err)
                return nil, err
            }
        }

        cfg.FileData = configFolderPath + "/" + "data.txt"
        cfg.FileRemove = configFolderPath + "/" + "delete.txt"
    }

    file, err := os.OpenFile(cfg.FileData, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
    if err != nil {
        fmt.Println("Error opening file data:", err)
        return nil, err
    }

    fileDelete, err := os.OpenFile(cfg.FileRemove, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
    if err != nil {
        fmt.Println("Error opening file delete:", err)
        return nil, err
    }

    return &Engine{
        data:       make(map[string]int64),
        file:       file,
        fileDelete: fileDelete,
        mu:         sync.Mutex{},
        muDelete:   sync.Mutex{},
    }, nil
}

func getConfigFolder() (string, error) {
    currentUser, err := user.Current()
    if err != nil {
        return "", err
    }

    homeDir := currentUser.HomeDir
    configFolder := ".config/keyvaluedb"
    configFolderPath := filepath.Join(homeDir, configFolder)
    return configFolderPath, nil
}

We add a Config struct , this will help us for set file data and delete on testing, NewEngine function now receives a config struct as a parameter, if the file data and delete is not set ( empty strings default ), we use the current user path /.config/keyvaluedb to save data of the application,
we use the function getConfigFolder to get the path of the current user, on NewEngine we check if that folder exists and if not we create it, after that we set the path of the files using the config struct.

We need to update the tests to check this.


var cfg = Config{
    FileData:   "data.txt",
    FileRemove: "delete.txt",
}

func Test_SetGetKeyValue(t *testing.T) {
    e, _ := NewEngine(cfg)
    e.Set("test", "data")
    e.Set("foo", "bar")
    value, err := e.Get("foo")
    if err != nil {
        t.Error(err)
    }
    if value != "bar" {
        t.Error("value should be bar")
    }

    _, err = e.Get("notfound")
    if err == nil {
        t.Error("should return error")
    }
}

We need to update all the calls to NewEngine in order use the cfg variable.

Update main.go

e, err = NewEngine(Config{})

In this case we pass a empty config struct, we are going to use that for the http server.

Run the server and save some data

go run .
curl -X POST -H "Content-Type: application/json" -d '{"key": "mykey", "value": "bar"}' http://localhost:8080/set

After that we can check the data.txt file on the path of the current user

cat ~/.config/keyvaluedb/data.txt

Full code github
https://github.com/ernesto27/system-design/tree/master/keyvaluedb/tutorial

Conclusion

In this tutorial we finished a very simple implementation of a key value store, although you must use in production a real and stable Database, is a good exercise to understand how things are made from scatch and understand concepts that help us how to choose our next database for some project.

we used a lot of concepts like concurrency, mutex, file write/read, http server, etc that would be very useful in other projects.

Create a golang cli using openAI-chatgpt API

Ernesto Ponce — Sun, 05 Mar 2023 06:54:24 +0000

In this tutorial we are going to create a simple golang command line program that uses the openAI API to recreate a similar funcionality of the chatgpt web.
https://chat.openai.com/chat

We will see this

OpenAI apikey setup
Setup base golang project
Get input from terminal
Send question to openAI API
Show the results

This is the final result of the app

This gif was created using the vhs amazing tool.
https://github.com/charmbracelet/vhs

The funcionality is pretty simple.

Execute a binary program.
Ask a question, then press enter.
The program send a questions to the openAI API and show the response, all on the terminal.

OpenAI apikey

In order to complete this tutorial you must obtain a apikey,
go to openAI and get a new one

https://platform.openai.com/account/api-keys

Be sure that you save the apikey in some place, because they will show you it only once time.

Setup project

Go to your terminal, and run this command in order to create a folder and go inside

mkdir chatgptcli && cd chatgptcli

After that, init a go module

go mod init chatgptcli

Create a main.go file

touch main.go

Get input from on terminal

In the main.go file add this code.

func main() {
    reader := bufio.NewReader(os.Stdin)
    fmt.Print("Ask to chat gpt: ")
    text, err := reader.ReadString('\n')
    if err != nil {
        panic(err)
    }

    fmt.Println("User input:", text)
}

We created a reader object passing the os.Stdin as parameter in order to obtain the input from the command line.
After we use the ReadString that reads the input until a break line ( Enter key pressed ) is found, this is defined by the parameter '\n'.

If we have an error we finish the programa using a call to a panic function, this is not the most elegant way to manage this but for the purpose of the tutorial is enough.

After that we only print on console the user input text.

Send question to openAI API

We will use this great library to connect to the openAPI API
https://github.com/sashabaranov/go-openai

That bring us a nice interface to connect to the api without the need to know all the params, configurations, etc, that the openAI required for work.

This is the official documentation if you want to take a look.
https://platform.openai.com/docs/api-reference/chat

Add the library to the project

go get github.com/sashabaranov/go-openai

Add this code to main function on file main.go

func main() {
        // previous code
        // [...] 

    c := openai.NewClient("youropenaitoken"))

    resp, err := c.CreateChatCompletion(
        context.Background(),
        openai.ChatCompletionRequest{
            Model: openai.GPT3Dot5Turbo,
            Messages: []openai.ChatCompletionMessage{
                {
                    Role:    "user",
                    Content: text,
                },
            },
        },
    )

    if err != nil {
        panic(err)
    }

    fmt.Println(resp.Choices[0].Message.Content)
    fmt.Println()

}

This block of code create a newClient using the openAI token, in this project we hardcode this value but if you use this in a project ṕlease consider to use an environment variable instead of this.

We use the method createChatCompletion and the model openai.GPT3Dot5Turbo as our configuration.

The official definition for this model is the following

A set of models that improve on GPT-3 and can understand as well as generate natural language or code

There are more options to choose

DALL-E
Whisper
Codex
etc

https://platform.openai.com/docs/models/overview

All of them or at least most of them are available to use with the library.

Run program with this command

go run main.go

That works fine but we have a problem, after the response for the API appears in the terminal, the programs finish and if we want to ask another question we have to open the program again.
We can change this behaviour and prevent that, adding a little change.

Add a "for" that wraps all main code.

func main() {

    for {
        reader := bufio.NewReader(os.Stdin)
        fmt.Print("Ask to chat gpt: ")
        text, err := reader.ReadString('\n')
        if err != nil {
            fmt.Println("Error reading input:", err)
            return
        }

        fmt.Println("Input text:", text)

        c := openai.NewClient("youropenaitoken")

        resp, err := c.CreateChatCompletion(
            context.Background(),
            openai.ChatCompletionRequest{
                Model: openai.GPT3Dot5Turbo,
                Messages: []openai.ChatCompletionMessage{
                    {
                        Role:    "user",
                        Content: text,
                    },
                },
            },
        )

        if err != nil {
            panic(err)
        }

        fmt.Println(resp.Choices[0].Message.Content)
        fmt.Println()
    }
}

The for loops run infinitely ( like a "while true" in other languages ) and the promp to enter text will appears after see the results from the API, with than change now we have to close the program using CTRL+C.

So, that is all for now, you can see all the code in this gist.

https://gist.github.com/ernesto27/6aadbb2ed8169294c375505733381f16

This program works fine, but we can make some improvements for future tutorials, for example.

The conversation starts from scratch always, i am not sure if is possible to mantain a conversation with the API.
We can improve how the text looks in the terminal.
We can configure the program to use a env variable instead of hardcode token.

You also can find a more complete version of this program on this repository.
https://github.com/ernesto27/chatgpt-cli