1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
package jobCheckin
import (
"app/common"
"encoding/json"
client "github.com/influxdata/influxdb1-client/v2"
"log"
"net/http"
"os"
"regexp"
"strings"
)
type Job struct {
Source string
}
// Validates the incomming request and writes an appropriate response.
// 405: when not POST request
// 401: when body is not given, or 'source' field missing
// 500: failed to complete
// 202: successfully complete
func jobCheckinHandler(w http.ResponseWriter, r *http.Request) {
log.Println("New request ===== ")
if r.Method != "POST" {
common.JsonRespond(&w, http.StatusMethodNotAllowed, "This end-point only accepts POST requests.")
return
}
var j Job
err := json.NewDecoder(r.Body).Decode(&j)
if err != nil {
common.JsonRespond(&w, http.StatusBadRequest, err.Error())
return
}
if j.Source == "" {
common.JsonRespond(&w, http.StatusBadRequest, "no 'source' field in body. Try sending a json such as {'source': 'job_name'} in the body.")
return
}
log.Println("source:", j.Source)
measurement, err := chekinJob(j.Source)
if err != nil {
common.JsonRespond(&w, http.StatusInternalServerError, "Error while writing to Influx.")
return
}
common.JsonRespond(&w, http.StatusAccepted, measurement)
return
}
// Write job check-in to influx.
func chekinJob(jobName string) (string, error) {
jobNameClean, err := sanitizeJobName(jobName)
if err != nil {
return "", err
}
// Make client
c, err := client.NewHTTPClient(client.HTTPConfig{
Addr: "http://" + os.Getenv("INFLUX_HOST") + ":" + os.Getenv("INFLUX_PORT"),
})
if err != nil {
return "", err
}
defer c.Close()
// Create a new point batch
bp, _ := client.NewBatchPoints(client.BatchPointsConfig{
Database: os.Getenv("INFLUX_DB"),
Precision: "s",
RetentionPolicy: os.Getenv("INFLUX_RP"),
})
measurement := jobNameClean + "_checkins"
// Create a point and add to batch
tags := map[string]string{}
fields := map[string]interface{}{"check-in": 1}
pt, err := client.NewPoint(measurement, tags, fields)
if err != nil {
return "", err
}
bp.AddPoint(pt)
// Write the batch
err = c.Write(bp)
if err != nil {
return "", err
}
return measurement, nil
}
// Sanitize jobname - we only want letters, numbers, and underscores
func sanitizeJobName(jobName string) (string, error) {
// remove any non-alphanumeric chars, and non-space chars
reg1, err := regexp.Compile("[^ _a-zA-Z0-9]+")
if err != nil {
return "", nil
}
jobName = reg1.ReplaceAllString(jobName, "")
// replace spaces with _
reg2, err := regexp.Compile("[ ]+")
if err != nil {
return "", nil
}
jobName = reg2.ReplaceAllString(jobName, "_")
return strings.ToLower(jobName), nil
}
// Every new endpoint/package MUST implement this function like so...
func MakeHTTPHandler(endpoint string) http.Handler {
return common.MakeHTTPHandler(endpoint, jobCheckinHandler)
}
|
