Understand Golang channels and how to monitor with Grafana (Part 2/2)
Tags: go, grafana, prometheus, concurrency
Published: 2019-12-09 19:56:48

---
series: Understand Golang channels and how to monitor with Grafana.
---

In the first part, we talked about what are go channels and how they work and how to implement them in a scalable way.

In this part, I'll show you how to monitor them and get some useful metrics like processing time per worker. How many jobs running per minute. And some other stuff

## Stack

let's take a small brief of the tools we will use to do the job.

- [Prometheus](https://prometheus.io/): is an open-source time-series metric system. also, it supports alerts but it's not our topic for today.

- [Grafana](https://grafana.com/): is an open-source analytics & monitor solution that supports many data sources and we will use it with Prometheus to display the metric we fire from our Go app.



## Prometheus

first, we need to define the metrics in our app and register them in Prometheus package. but let's define the types of the metric and what is the purpose of each one.

- **Counter:** a number that can only increase or reset to zero. like how many emails you sent. it can't be at some time 5 then became 3. it can only increase.

- **Gauge:** represents a single numerical value that can arbitrarily go up and down. like memory usage or concurrent users and for our case it's running workers and running jobs. because these numbers go up and down.

- **Histogram:** in just an observer. based on the samples it gives you the average. like response time, response size, request duration.

- **Summary:** It's like a histogram but also it counts the total observations and a sum of all observed values. 

*I'm not sure I understand summary well. but anyway we don't need it in this article*


```go
// prometheus.go

package queue

var (
    JobsProcessed  *prometheus.CounterVec
    RunningJobs    *prometheus.GaugeVec
    ProcessingTime *prometheus.HistogramVec
    RunningWorkers *prometheus.GaugeVec
)

var collectorContainer []prometheus.Collector

//InitPrometheus ... initalize prometheus
func InitPrometheus() {
    prometheus.MustRegister(collectorContainer...)
}

//PushRegister ... Push collectores to prometheus before inializing
func PushRegister(c ...prometheus.Collector) {
    collectorContainer = append(collectorContainer, c...)
}

func InitMetrics() {
    JobsProcessed = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Namespace: "worker",
            Subsystem: "jobs",
            Name:      "processed_total",
            Help:      "Total number of jobs processed by the workers",
        },
        []string{"worker_id", "type"},
    )

    RunningJobs = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Namespace: "worker",
            Subsystem: "jobs",
            Name:      "running",
            Help:      "Number of jobs inflight",
        },
        []string{"type"},
    )

    RunningWorkers = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Namespace: "worker",
            Subsystem: "workers",
            Name:      "running",
            Help:      "Number of workers inflight",
        },
        []string{"type"},
    )

    ProcessingTime = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Namespace: "worker",
            Subsystem: "jobs",
            Name:      "process_time_seconds",
            Help:      "Amount of time spent processing jobs",
        },
        []string{"worker_id", "type"},
    )

    metrics.PushRegister(ProcessingTime, RunningJobs, JobsProcessed, RunningWorkers)
}
      
```

after defining the metrics we need to run our endpoint that Prometheus will call to get the metrics data
```go
r.Handle("GET", "/metrics", gin.WrapH(promhttp.Handler()))
```

Now if we tried to make some calls 
```go
JobsProcessed.WithLabelValues("Worker-1", "ahmedash.com").Inc()
JobsProcessed.WithLabelValues("Worker-1", "ahmedash.com").Inc()

JobsProcessed.WithLabelValues("Worker-2", "ahmedash.com").Inc()
JobsProcessed.WithLabelValues("Worker-2", "ahmedash.com").Inc()
JobsProcessed.WithLabelValues("Worker-2", "ahmedash.com").Inc()
```

the result when you hit your Prometheus endpoint `http://localhost/metrics`

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575916507/dev.to/prometheus-metrics.png">

and if you opened Prometheus dashboard you will find the chart like below
<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575916585/dev.to/prometheus-dashboard.png">

this is the way of how you can work with Prometheus and send the data to it and check it's dashboard for simple visualization.

but as you can see the dashboard of Prometheus is not that fancy, and we need to create a nice dashboard so it's easy for anyone to understand what is going on with the system. and Grafana can tackle this job very well.


# Push the metric from Go to Prometheus 

From the [first part](https://dev.to/ahmedash95/understand-golang-channels-and-how-to-monitor-with-grafana-154), we will modify the code to send the data we need for the metrics

**Follow arrows ⬅️**
```go
func (d *Dispatcher) Run() {
    for i := 0; i < d.maxWorkers; i++ {
        
        // increase the number of running workers
        RunningWorkers.WithLabelValues("Emails").Inc() ⬅️⬅️⬅️⬅️⬅️⬅️⬅️⬅️

        worker := NewWorker(d.WorkerPool)
        worker.Start()
        d.Workers = append(d.Workers, worker)
    }

    go d.dispatch()
}

```

With every job. we need to increase the jobs in queue metric
```go

func (d *Dispatcher) dispatch() {
    for {
        select {
        case job := <-JobQueue:
            
            // Increase running jobs Gauge
            RunningJobs.WithLabelValues("Emails").Inc() ️⬅️⬅️⬅️⬅️⬅️⬅️⬅️

            go func(job Queuable) {
                jobChannel := <-d.WorkerPool
                jobChannel <- job
            }(job)
        }
    }
}
```

- First metric: stores the number of jobs processed per worker
- Second metric: decreases the number of current jobs running from the previous code snippet 
- Third metric: stores the processing time for each job

```go
func (w Worker) Start() {
    go func() {
        for {
            w.WorkerPool <- w.JobChannel

            select {
            case job := <-w.JobChannel:
                startTime := time.Now()

                // track the total number of jobs processed by the worker
                JobsProcessed.WithLabelValues(w.Name, "Emails").Inc() ️⬅️⬅️⬅️⬅️⬅️⬅️⬅️
                if err := job.Handle(); err != nil {
                    log.Fatal("Error in job: %s", err.Error())
                }
                // Decrease the number of running jobs once we finish
                RunningJobs.WithLabelValues("Emails").Dec() ️⬅️⬅️⬅️⬅️⬅️⬅️⬅️
                
                // ⬇️ Register the proccesing time in the Histogram ⬇️
                ProcessingTime.WithLabelValues(w.Name, "Emails").Observe(time.Now().Sub(startTime).Seconds()) ️⬅️⬅️⬅️⬅️⬅️⬅️⬅️
            }
        }
    }()
}
```


# Grafana

Now we have everything ready and our app is working as expected. the last step we need is just to have a beautiful dashboard with Grafana.

from Grafana dashboard we need to add Prometheus as a data source 

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575918893/dev.to/granfa-add-prometheus.png">

then from HOME at the top bar create **New** dashboard

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575918986/dev.to/grafana-add-dashobard.png">

Then we need to add a Singlestat for counters & gauge metrics

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919059/dev.to/add-singleslat.png">

Now:
1. Let's choose Prometheus from data-sources
2. then search for the metric name
3. have the full right name of the metric

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919169/dev.to/define-singlestat-metric.png">


Then from **Options** tab choose if it's current or total. in our case it's current

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919302/dev.to/options-current.png">

Finally from **General** tab rename the component

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919355/dev.to/general-title.png">


**Also**, you can control the size of each component in the panel
<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919420/dev.to/panel-size.png">


Let's make another one for the running workers with `worker_workers_running` metric name 

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919482/dev.to/2panels.png">

then let's make a graph panel to show how many jobs per minute

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919549/dev.to/2panels-w-graph.png">

Now you can make some requests to the app and see the data in realtime. just make sure to enable the auto-refresh.

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919613/dev.to/grafana-refresh.png">

Now we have the first part of our dashboard 
<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919663/dev.to/graph-running.png">

The second part is just about adding a graph to show the processing duration for each worker

for this metric, the value would be `worker_jobs_process_time_seconds_sum / worker_jobs_process_time_seconds_count`

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919814/dev.to/duration-metric.png">

and then make sure we display the values in milliseconds

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575919916/dev.to/define-unit.png">

and the final result is

<img src="https://res.cloudinary.com/ahmedash/image/upload/v1575920209/dev.to/final.gif">

Now we have a dashboard shows us the running workers, how many jobs in queues. jobs processing duration. also, easily we can see if there are any spikes with our jobs or the system works as we expect from him to be.


# Conclusion

In this article including the two parts. we learned how to 
- create channels
- dispatcher/worker pattern
- how to create&kill workers
- push metrics and display them in a very nice dashboard


**Things we didn't cover**
- We don't deal with failed jobs. so if an error happens to any job we don't have a way to re-try or even store them for later
- No data persistence. so if the app restarts we will lose all jobs because they are in memory.
- We didn't show how to make this work with a third-party messaging/queuing system like Redis or RabbitMQ.

# Source code
This is a source code contains all the code we went through. should be easy to run as it's dockerized.

https://github.com/ahmedash95/go-channels-demo

# References
These 2 articles helped me a lot to understand how I can build powerful apps that support async jobs and monitor them. only my article combines knowledge from.

- [Handling 1 Million Requests per Minute with Golang](https://medium.com/smsjunk/handling-1-million-requests-per-minute-with-golang-f70ac505fcaa)
- [Monitoring Go Applications With Prometheus](https://scot.coffee/2018/12/monitoring-go-applications-with-prometheus/)