-
Notifications
You must be signed in to change notification settings - Fork 23
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Pawel Szczodruch
committed
Apr 7, 2021
1 parent
e27f702
commit d642ab0
Showing
6 changed files
with
232 additions
and
10 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| package rollbar | ||
|
|
||
| import "sync" | ||
|
|
||
| // NewQueue returns a new queue with the given initial size. | ||
| func NewQueue(size int) *Queue { | ||
| return &Queue{ | ||
| nodes: make([]interface{}, size), | ||
| size: size, | ||
| } | ||
| } | ||
|
|
||
| // Queue is a basic FIFO queue based on a circular list that resizes as needed. | ||
| type Queue struct { | ||
| nodes []interface{} | ||
| size int | ||
| head int | ||
| tail int | ||
| count int | ||
|
|
||
| lock sync.RWMutex | ||
| } | ||
|
|
||
| // Push adds a node to the queue. | ||
| func (q *Queue) Push(n interface{}) { | ||
| q.lock.Lock() | ||
| defer q.lock.Unlock() | ||
| if q.head == q.tail && q.count > 0 { | ||
| nodes := make([]interface{}, len(q.nodes)+q.size) | ||
| copy(nodes, q.nodes[q.head:]) | ||
| copy(nodes[len(q.nodes)-q.head:], q.nodes[:q.head]) | ||
| q.head = 0 | ||
| q.tail = len(q.nodes) | ||
| q.nodes = nodes | ||
| } | ||
| q.nodes[q.tail] = n | ||
| q.tail = (q.tail + 1) % len(q.nodes) | ||
| q.count++ | ||
| } | ||
|
|
||
| // Pop removes and returns a node from the queue in first to last order. | ||
| func (q *Queue) Pop() interface{} { | ||
| q.lock.Lock() | ||
| defer q.lock.Unlock() | ||
| if q.count == 0 { | ||
| return nil | ||
| } | ||
| node := q.nodes[q.head] | ||
| q.head = (q.head + 1) % len(q.nodes) | ||
| q.count-- | ||
| return node | ||
| } | ||
|
|
||
| // Items returns all populated (non nil) items | ||
| func (q *Queue) Items() []interface{} { | ||
| q.lock.RLock() | ||
| defer q.lock.RUnlock() | ||
| return q.nodes[:q.count] | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| package rollbar | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "io" | ||
| "log" | ||
| "net/http" | ||
| "os" | ||
| "time" | ||
| ) | ||
|
|
||
| // Telemetry struct contains writer (for logs) and round tripper (for http client) and enables to queue the events | ||
| type Telemetry struct { | ||
| Writer io.Writer | ||
| Proxied http.RoundTripper | ||
| Queue *Queue | ||
| } | ||
|
|
||
| // Write is the writer for telemetry logs | ||
| func (t *Telemetry) Write(p []byte) (int, error) { | ||
| telemetryData := t.populateLoggerBody(p) | ||
| t.Queue.Push(telemetryData) | ||
| return t.Writer.Write(p) | ||
| } | ||
|
|
||
| // RoundTrip implements RoundTrip in http.RoundTripper | ||
| func (t *Telemetry) RoundTrip(req *http.Request) (res *http.Response, e error) { | ||
|
|
||
| // Send the request, get the response (or the error) | ||
| res, e = t.Proxied.RoundTrip(req) | ||
| if e != nil { | ||
| fmt.Printf("Error: %v", e) | ||
| } | ||
| telemetryData := t.populateTransporterBody(req, res) | ||
| t.Queue.Push(telemetryData) | ||
| return | ||
| } | ||
|
|
||
| func (t *Telemetry) populateLoggerBody(p []byte) map[string]interface{} { | ||
| var data = map[string]interface{}{} | ||
| message := map[string]interface{}{"message": string(p)} | ||
| data["body"] = message | ||
| data["source"] = "client" | ||
| data["timestamp_ms"] = time.Now().UnixNano() / int64(time.Millisecond) | ||
| data["type"] = "log" | ||
| data["level"] = "log" | ||
| return data | ||
| } | ||
|
|
||
| func (t *Telemetry) populateTransporterBody(req *http.Request, res *http.Response) map[string]interface{} { | ||
| var data = map[string]interface{}{} | ||
| var dataBody = map[string]interface{}{} | ||
| var dataHeaders = map[string][]string{} | ||
| dataBody["status_code"] = nil | ||
| data["level"] = "info" | ||
| if res != nil { | ||
| dataBody["status_code"] = res.StatusCode | ||
| if res.StatusCode >= http.StatusInternalServerError { | ||
| data["level"] = "critical" | ||
| } else if res.StatusCode >= http.StatusBadRequest { | ||
| data["level"] = "error" | ||
| } | ||
| } | ||
| dataBody["url"] = req.URL.Scheme + "://" + req.Host + req.URL.Path | ||
| dataBody["method"] = req.Method | ||
| dataBody["subtype"] = "http" | ||
|
|
||
| for k, v := range req.Header { | ||
| dataHeaders[k] = v | ||
| } | ||
| dataBody["request_headers"] = dataHeaders | ||
|
|
||
| data["body"] = dataBody | ||
| data["source"] = "client" | ||
| data["timestamp_ms"] = time.Now().UnixNano() / int64(time.Millisecond) | ||
| data["type"] = "network" | ||
| return data | ||
| } | ||
|
|
||
| // GetQueueItems gets all the items from the queue | ||
| func (t *Telemetry) GetQueueItems() []interface{} { | ||
| return t.Queue.Items() | ||
| } | ||
|
|
||
| // OptionFunc is the pointer to the optional parameter function | ||
| type OptionFunc func(*Telemetry) | ||
|
|
||
| // WithCustomTransporter sets the custom transporter | ||
| func WithCustomTransporter(t http.RoundTripper) OptionFunc { | ||
| return func(f *Telemetry) { | ||
| f.Proxied = t | ||
| } | ||
| } | ||
|
|
||
| // WithCustomQueueSize initializes the queue with a custom size | ||
| func WithCustomQueueSize(size int) OptionFunc { | ||
| return func(f *Telemetry) { | ||
| f.Queue = NewQueue(size) | ||
| } | ||
| } | ||
|
|
||
| // NewTelemetry initializes telemetry object | ||
| func NewTelemetry(options ...OptionFunc) *Telemetry { | ||
| res := &Telemetry{ | ||
| Proxied: http.DefaultTransport, | ||
| Queue: NewQueue(50), | ||
| Writer: os.Stdout, | ||
| } | ||
| for _, opt := range options { | ||
| opt(res) | ||
| } | ||
|
|
||
| log.SetOutput(res) | ||
| http.DefaultClient = &http.Client{Transport: res} | ||
| return res | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters