Forem: Arunachalam Lakshmanan

Simple Go App to post messages to AWS SQS

Arunachalam Lakshmanan — Tue, 08 Jun 2021 22:49:37 +0000

AWS SQS: Amazon Simple Queue Service (SQS) is a fully managed message queuing service that enables you to decouple and scale microservices, distributed systems, and serverless applications.

Some time we may need to send messages to AWS SQS along with message meta data. I use a simple program written in golang to post a message.

As we do with all AWS programmatic access, we need to create session

sess := session.Must(session.NewSessionWithOptions(session.Options{
        SharedConfigState: session.SharedConfigEnable,
    }))

Then we need to get the SQS queue url from the queue name

func GetQueueURL(sess *session.Session, queue *string) (*sqs.GetQueueUrlOutput, error) {
    // Create an SQS service client
    svc := sqs.New(sess)

    result, err := svc.GetQueueUrl(&sqs.GetQueueUrlInput{
        QueueName: queue,
    })
    if err != nil {
        return nil, err
    }

    return result, nil
}

You can attach metadata to the message while posting it to the SQS queue

func SendMsg(sess *session.Session, data string, id string, msgType string, queueURL *string) error {
    // Create an SQS service client
    svc := sqs.New(sess)
    var err error

    _, err = svc.SendMessage(&sqs.SendMessageInput{
        MessageAttributes: map[string]*sqs.MessageAttributeValue{
            "msgType": {
                DataType:    aws.String("String"),
                StringValue: aws.String(msgType),
            },
            "id": {
                DataType:    aws.String("String"),
                StringValue: aws.String(id),
            },
        },
        MessageBody: aws.String(data),
        QueueUrl:    queueURL,
    })

    if err != nil {
        return err
    }

    return nil
}

The whole code can be found at github repo https://github.com/arunx2/golang-samples/blob/main/publish-to-sqs/main.go

Simple Slack Notification with golang

Arunachalam Lakshmanan — Mon, 20 Jul 2020 02:43:05 +0000

A slack notification is useful when

A long-running job is started
When some critical error
A specific request comes through the process as cache invalidate, critical delete request, etc...

However slack provides simple webhooks to notify a channel, it has more features like action, workflow, poll, etc... So the slack go libraries need to comply with these features and that makes the library heavy by bringing internal dependencies. We can write a small tool in Go just for webhooks. It just uses standard libraries.

Usage

Create a slack client with an optional user name one per process. The SendSlackNotification() can send a simple notification to the given channel.

Options: Username - the name of the bot in a slack message
Timeout - default would be 5 seconds
Icon_emoji - the name of the icon image, otherwise it is a default image of incoming webhook.

Sample

    sc := SlackClient{
        WebHookUrl: "https://WEB_HOOK_URL",
        UserName:   "USER_NAME",
        Channel:    "CHANNEL_NAME",
    }
    sr := SimpleSlackRequest{
        Text:      "This is test message",
        IconEmoji: ":ghost:",
    }
    err := sc.SendSlackNotification(sr)
    if err != nil {
        log.Fatal(err)
    }

    To send a notification with status (slack attachments)
    sr := SlackJobNotification{
        Text:  "This is attachment message",
        Details: "details of the jobs",
        Color: "warning",
        IconEmoji: ":hammer_and_wrench",
    }
    err := sc.SendJobNotification(sr)
    if err != nil {
        log.Fatal(err)
    }

Source

package notification

import (
    "bytes"
    "encoding/json"
    "errors"
    "net/http"
    "strconv"
    "time"
)

const DefaultSlackTimeout = 5 * time.Second

type SlackClient struct {
    WebHookUrl string
    UserName   string
    Channel    string
    TimeOut    time.Duration
}

type SimpleSlackRequest struct {
    Text      string
    IconEmoji string
}

type SlackJobNotification struct {
    Color     string
    IconEmoji string
    Details   string
    Text      string
}

type SlackMessage struct {
    Username    string       `json:"username,omitempty"`
    IconEmoji   string       `json:"icon_emoji,omitempty"`
    Channel     string       `json:"channel,omitempty"`
    Text        string       `json:"text,omitempty"`
    Attachments []Attachment `json:"attachments,omitempty"`
}

type Attachment struct {
    Color         string `json:"color,omitempty"`
    Fallback      string `json:"fallback,omitempty"`
    CallbackID    string `json:"callback_id,omitempty"`
    ID            int    `json:"id,omitempty"`
    AuthorID      string `json:"author_id,omitempty"`
    AuthorName    string `json:"author_name,omitempty"`
    AuthorSubname string `json:"author_subname,omitempty"`
    AuthorLink    string `json:"author_link,omitempty"`
    AuthorIcon    string `json:"author_icon,omitempty"`
    Title         string `json:"title,omitempty"`
    TitleLink     string `json:"title_link,omitempty"`
    Pretext       string `json:"pretext,omitempty"`
    Text          string `json:"text,omitempty"`
    ImageURL      string `json:"image_url,omitempty"`
    ThumbURL      string `json:"thumb_url,omitempty"`
    // Fields and actions are not defined.
    MarkdownIn []string    `json:"mrkdwn_in,omitempty"`
    Ts         json.Number `json:"ts,omitempty"`
}

// SendSlackNotification will post to an 'Incoming Webook' url setup in Slack Apps. It accepts
// some text and the slack channel is saved within Slack.
func (sc SlackClient) SendSlackNotification(sr SimpleSlackRequest) error {
    slackRequest := SlackMessage{
        Text:      sr.Text,
        Username:  sc.UserName,
        IconEmoji: sr.IconEmoji,
        Channel:   sc.Channel,
    }
    return sc.sendHttpRequest(slackRequest)
}

func (sc SlackClient) SendJobNotification(job SlackJobNotification) error {
    attachment := Attachment{
        Color: job.Color,
        Text:  job.Details,
        Ts:    json.Number(strconv.FormatInt(time.Now().Unix(), 10)),
    }
    slackRequest := SlackMessage{
        Text:        job.Text,
        Username:    sc.UserName,
        IconEmoji:   job.IconEmoji,
        Channel:     sc.Channel,
        Attachments: []Attachment{attachment},
    }
    return sc.sendHttpRequest(slackRequest)
}

func (sc SlackClient) SendError(message string, options ...string) (err error) {
    return sc.funcName("danger", message, options)
}

func (sc SlackClient) SendInfo(message string, options ...string) (err error) {
    return sc.funcName("good", message, options)
}

func (sc SlackClient) SendWarning(message string, options ...string) (err error) {
    return sc.funcName("warning", message, options)
}

func (sc SlackClient) funcName(color string, message string, options []string) error {
    emoji := ":hammer_and_wrench"
    if len(options) > 0 {
        emoji = options[0]
    }
    sjn := SlackJobNotification{
        Color:     color,
        IconEmoji: emoji,
        Details:   message,
    }
    return sc.SendJobNotification(sjn)
}
func (sc SlackClient) sendHttpRequest(slackRequest SlackMessage) error {
    slackBody, _ := json.Marshal(slackRequest)
    req, err := http.NewRequest(http.MethodPost, sc.WebHookUrl, bytes.NewBuffer(slackBody))
    if err != nil {
        return err
    }
    req.Header.Add("Content-Type", "application/json")
    if sc.TimeOut == 0 {
        sc.TimeOut = DefaultSlackTimeout
    }
    client := &http.Client{Timeout: sc.TimeOut}
    resp, err := client.Do(req)
    if err != nil {
        return err
    }

    buf := new(bytes.Buffer)
    _, err = buf.ReadFrom(resp.Body)
    if err != nil {
        return err
    }
    if buf.String() != "ok" {
        return errors.New("Non-ok response returned from Slack")
    }
    return nil
}

Working with application logs

Arunachalam Lakshmanan — Thu, 16 Jul 2020 02:54:09 +0000

A small tutorial to work with application logs and analyze them with Elastic and Kibana. In a microservices application architecture, it is important to analyze the logs to understand the health of the application. We can extract several stats from the logs when a log is a strucApplication logs can be generated in multiple ways. They can be either simple logs or print to console (with no time stamps).

Collecting logs

Log files - Most applications default behaviour is to write the logs on the file system. They may rolled up by date or size. A simple file beat process in the server can export the logs to either logstash or fluentd or directly to elastic.
- Advantages:
  - Simple file system
  - Backups available as long as the file is purged.
  - No code changes required.
- Disadvantages:
  - As many as file exporter processes required to export the logs.
  - May miss console outputs.
  - In container environment, we need to create a volume for logs and mount them when spinning a container.
Message brokers - When writing logs into file system, there is a possibility that we need to clean up the disk space regularly. It can be cumbersome in containers (creating a volume and mount to containers). There are language specific log appenders available to push the logs directly from application.
- Advantages:
  - No file system
  - Log appenders can asynchronysly push logs to message brokers
  - Messages can be dropped after TTL
  - If the upstream processes are not accessible, logs won't be lost
- Disadvantages
  - Need infrastructure for message brokers.
Log drivers - This is more suitable option for legacy application where there are limitted accessability to code or change in application is forbidden.
- Advantages:
  - No code change required.
  - Everything that application emits including console prints.
  - Several log drivers comes by default with docker runtime.
  - No additional infrastructure.
- Disadvantages:
  - There may be data lose with some protocols.
  - Some log drivers can cause application hung when the upstream is not processing the data correctly or if not designed correctly.

Collecting logs via message brokers is most reliable option when considering failures in each layer even we need infrastructure for brokers.

Best practices

Use JSON: Json format logs can be helpful rather than unstructured data.
One log per function: Try to write one log function per exit point of the function.
Check for isDebugEnabled: String concatenation is super expensive in any language. Do you message construction after checking the log level as most of the time the debug would be disabled.
In microservices environment, keep the kafka topic name with some prefix. Just one process is sufficient in processing multiple topics.
Keep index templates and enable index policies in Kibana. If logs needs to be preserved for long period of time, consider moving them to cold storage after certain period. We can still search them but stored in inexpensive storage.

Sample logstash config

Sample log stash configuration consuming messages from multiple topics of kafka and ingest into individual indices in elastic search.

input {
    kafka {
        codec => "json"
        bootstrap_servers => "broker1:9092,broker2:9092"
        topics_pattern => "prefix-.*-logs"
        group_id => "consumer_group_id"
        auto_offset_reset => "earliest"
        decorate_events => true
        metadata_max_age_ms => 60000
        heartbeat_interval_ms => "250000"
        session_timeout_ms => "300000"
        poll_timeout_ms => "30000"
        consumer_threads => 6
    }
}

filter {
 kv {
     #default source field is message
     #source => "message"
     field_split_pattern => "([,\s]+)(?=\w+\s*\=)"
     transform_key => "lowercase"
     recursive => "true"
     trim_value => "<>\[\],"

}
grok {
    match => [ "message", "%{GREEDYDATA}" ]
}

mutate {
    #some languages emits json logs with double quotes. we need to change them to
    #single quote before parsing as json.
    gsub => [
            "msg", '"{', "'{",
            "msg", '}"', "}'"
        ]
}
json {
    source => "msg"
}
date {
    match => ["[@metadata][kafka][timestamp]", "UNIX_MS"]
    target =>  "@timestamp"
   }
ruby {
    code => "event.set('[@metadata][kafka][lc_topic]', event.get('[@metadata][kafka][topic]').split(/(?=[A-Z])/).map{|x| x.downcase }.join('_') )"
  }
}

output {
    elasticsearch{
        index => "%{[@metadata][kafka][lc_topic]}-%{+YYYY.MM.dd}"
        hosts => ["http://host1:9200", "http://host2:9200"]
    }
    #stdout {
        #codec => rubydebug { metadata => true }
    #}
}