-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Initial cloudwatch implementation * Updated with `context` * Refactored signal handling slightly * Fixes for cloudwatch * Initial cloudwatch implementation * Updated with `context` * Fixes for cloudwatch * Factored out retry logic and LogMessage channel deduplication * More graceful recovery upon parse errors of the attribute cache, also: removing slack.token * Follow support for cloudwatch * README update for cloudwatch * Context only for queries, not handled elsewhere yet. * Implemented --before and --after for cloudwatch * Graceful cache fallback * Rename 'client' to 'logs' * Fixes and refactoring of follow requerying
- Loading branch information
1 parent
789fcff
commit 2510860
Showing
13 changed files
with
495 additions
and
99 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 was deleted.
Oops, something went wrong.
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,145 @@ | ||
| package cloudwatch | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/aws/aws-sdk-go/aws" | ||
| "github.com/aws/aws-sdk-go/aws/credentials" | ||
| "github.com/aws/aws-sdk-go/aws/session" | ||
| "github.com/aws/aws-sdk-go/service/cloudwatchlogs" | ||
| "github.com/egnyte/ax/pkg/backend/common" | ||
| ) | ||
|
|
||
| type CloudwatchClient struct { | ||
| logs *cloudwatchlogs.CloudWatchLogs | ||
| groupName string | ||
| } | ||
|
|
||
| func attemptParseJSON(str string) map[string]interface{} { | ||
| m := make(map[string]interface{}) | ||
| // Find start of JSON blob | ||
| startIdx := strings.Index(str, "{") | ||
| if startIdx == -1 { // If not found, fall back to dumping the whole thing into the "message" field | ||
| m["message"] = str | ||
| return m | ||
| } | ||
| err := json.Unmarshal([]byte(str[startIdx:]), &m) | ||
| if err != nil { | ||
| m["message"] = str | ||
| } | ||
| return m | ||
| } | ||
|
|
||
| func logEventToMessage(query common.Query, logEvent *cloudwatchlogs.FilteredLogEvent) common.LogMessage { | ||
| message := common.NewLogMessage() | ||
| message.ID = *logEvent.EventId | ||
| message.Timestamp = time.Unix((*logEvent.Timestamp)/1000, (*logEvent.Timestamp)%1000) | ||
| message.Attributes = common.Project(attemptParseJSON(*logEvent.Message), query.SelectFields) | ||
| return message | ||
| } | ||
|
|
||
| // https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/FilterAndPatternSyntax.html | ||
| func queryToFilterPattern(query common.Query) string { | ||
| filterParts := make([]string, 0) | ||
| for _, filter := range query.Filters { | ||
| filterParts = append(filterParts, fmt.Sprintf("($.%s %s \"%s\")", filter.FieldName, filter.Operator, filter.Value)) | ||
| } | ||
| var filterPattern string | ||
| if len(query.Filters) == 0 { | ||
| filterPattern = query.QueryString | ||
| } else { | ||
| filterPattern = fmt.Sprintf("%s { %s }", query.QueryString, strings.Join(filterParts, " && ")) | ||
| } | ||
|
|
||
| return strings.TrimSpace(filterPattern) | ||
| } | ||
|
|
||
| func (client *CloudwatchClient) readLogBatch(ctx context.Context, query common.Query) ([]common.LogMessage, error) { | ||
| var startTime, endTime *int64 = nil, nil | ||
| if query.After != nil { | ||
| startTimeVal := (*query.After).UnixNano() / int64(time.Millisecond) | ||
| startTime = &startTimeVal | ||
| } | ||
| if query.Before != nil { | ||
| endTimeVal := (*query.Before).UnixNano() / int64(time.Millisecond) | ||
| endTime = &endTimeVal | ||
| } | ||
| resp, err := client.logs.FilterLogEventsWithContext(ctx, &cloudwatchlogs.FilterLogEventsInput{ | ||
| LogGroupName: aws.String(client.groupName), | ||
| FilterPattern: aws.String(queryToFilterPattern(query)), | ||
| Limit: aws.Int64(int64(query.MaxResults)), | ||
| StartTime: startTime, | ||
| EndTime: endTime, | ||
| }) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| messages := make([]common.LogMessage, 0, 20) | ||
| for _, message := range resp.Events { | ||
| messages = append(messages, logEventToMessage(query, message)) | ||
| } | ||
| return messages, nil | ||
| } | ||
|
|
||
| func (client *CloudwatchClient) Query(ctx context.Context, query common.Query) <-chan common.LogMessage { | ||
| if query.Follow { | ||
| return common.ReQueryFollow(ctx, func() ([]common.LogMessage, error) { | ||
| return client.readLogBatch(ctx, query) | ||
| }) | ||
| } | ||
| resultChan := make(chan common.LogMessage) | ||
|
|
||
| go func() { | ||
| messages, err := client.readLogBatch(ctx, query) | ||
| if err != nil { | ||
| fmt.Printf("Error while fetching logs: %s\n", err) | ||
| close(resultChan) | ||
| return | ||
| } | ||
| for _, message := range messages { | ||
| resultChan <- message | ||
| } | ||
| close(resultChan) | ||
| }() | ||
|
|
||
| return resultChan | ||
| } | ||
|
|
||
| func (client *CloudwatchClient) ListGroups() ([]string, error) { | ||
| resp, err := client.logs.DescribeLogGroups(&cloudwatchlogs.DescribeLogGroupsInput{}) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| groupNames := make([]string, 0) | ||
| for _, stream := range resp.LogGroups { | ||
| groupNames = append(groupNames, *stream.LogGroupName) | ||
| } | ||
|
|
||
| return groupNames, err | ||
| } | ||
|
|
||
| func New(accessKey, accessSecretKey, region, groupName string) *CloudwatchClient { | ||
| sess, err := session.NewSession(&aws.Config{ | ||
| Region: aws.String(region), | ||
| Credentials: credentials.NewStaticCredentials(accessKey, accessSecretKey, ""), | ||
| }) | ||
|
|
||
| if err != nil { | ||
| fmt.Printf("Could not create AWS Session: %s\n", err) | ||
| return nil | ||
| } | ||
| logs := cloudwatchlogs.New(sess) | ||
|
|
||
| return &CloudwatchClient{ | ||
| logs: logs, | ||
| groupName: groupName, | ||
| } | ||
|
|
||
| } | ||
|
|
||
| var _ common.Client = &CloudwatchClient{} |
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,69 @@ | ||
| package cloudwatch | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/egnyte/ax/pkg/backend/common" | ||
| ) | ||
|
|
||
| func TestParsing(t *testing.T) { | ||
| msg := attemptParseJSON(`2017-09-27T09:01:01.245468966Z {"asctime": "2017-09-27 09:01:01,245", "created": 1506502861.2452097, "filename": "connectionpool.py", "funcName": "_make_request", "levelname": "DEBUG", "levelno": 10, "module": "connectionpool", "msecs": 245.2096939086914, "message": "http://localhost:None \"POST /v1.29/exec/1744fb9d8aa1ed1f94f729d4e0474251dfab9e0523385d42e77ea10acda53957/start HTTP/1.1\" 101 0", "name": "urllib3.connectionpool", "pathname": "/usr/local/lib/python3.6/site-packages/urllib3/connectionpool.py", "process": 5, "processName": "MainProcess", "relativeCreated": 2276.298999786377, "thread": 140018892404480, "threadName": "MainThread", "turbo_request_id": null, "user": null, "tid": 5, "source": "/usr/local/lib/python3.6/site-packages/urllib3/connectionpool.py:395", "client_id": null} | ||
| `) | ||
| if msg["filename"] != "connectionpool.py" { | ||
| t.Errorf("Parsed: %+v", msg) | ||
| t.Fail() | ||
| } | ||
| msg = attemptParseJSON(`{"asctime": "2017-09-27 09:01:01,245", "created": 1506502861.2452097, "filename": "connectionpool.py", "funcName": "_make_request", "levelname": "DEBUG", "levelno": 10, "module": "connectionpool", "msecs": 245.2096939086914, "message": "http://localhost:None \"POST /v1.29/exec/1744fb9d8aa1ed1f94f729d4e0474251dfab9e0523385d42e77ea10acda53957/start HTTP/1.1\" 101 0", "name": "urllib3.connectionpool", "pathname": "/usr/local/lib/python3.6/site-packages/urllib3/connectionpool.py", "process": 5, "processName": "MainProcess", "relativeCreated": 2276.298999786377, "thread": 140018892404480, "threadName": "MainThread", "turbo_request_id": null, "user": null, "tid": 5, "source": "/usr/local/lib/python3.6/site-packages/urllib3/connectionpool.py:395", "client_id": null} | ||
| `) | ||
| if msg["filename"] != "connectionpool.py" { | ||
| t.Errorf("Parsed: %+v", msg) | ||
| t.Fail() | ||
| } | ||
| } | ||
|
|
||
| func TestFilterGenerator(t *testing.T) { | ||
| output := queryToFilterPattern(common.Query{ | ||
| QueryString: "Test", | ||
| Filters: []common.QueryFilter{ | ||
| { | ||
| FieldName: "name", | ||
| Operator: "=", | ||
| Value: "zef", | ||
| }, | ||
| }, | ||
| }) | ||
| if output != `Test { ($.name = "zef") }` { | ||
| t.Fatal(output) | ||
| } | ||
|
|
||
| output = queryToFilterPattern(common.Query{ | ||
| Filters: []common.QueryFilter{ | ||
| { | ||
| FieldName: "name", | ||
| Operator: "=", | ||
| Value: "zef", | ||
| }, | ||
| }, | ||
| }) | ||
| if output != `{ ($.name = "zef") }` { | ||
| t.Fatal(output) | ||
| } | ||
|
|
||
| output = queryToFilterPattern(common.Query{ | ||
| Filters: []common.QueryFilter{ | ||
| { | ||
| FieldName: "name", | ||
| Operator: "=", | ||
| Value: "zef", | ||
| }, | ||
| { | ||
| FieldName: "age", | ||
| Operator: "=", | ||
| Value: "34", | ||
| }, | ||
| }, | ||
| }) | ||
| if output != `{ ($.name = "zef") && ($.age = "34") }` { | ||
| t.Fatal(output) | ||
| } | ||
| } |
Oops, something went wrong.