[Filebeat][okta] Fix okta pagination (#21797)

marc-gr · web-flow · commit 80d4209bc90d · 2020-10-16T09:01:34.000+02:00
* Fix okta pagination

* Use cursor storage
diff --git a/x-pack/filebeat/input/httpjson/date_cursor.go b/x-pack/filebeat/input/httpjson/date_cursor.go
@@ -41,7 +41,9 @@ func newDateCursorFromConfig(config config, log *logp.Logger) *dateCursor {
 	c.urlField = config.DateCursor.URLField
 	c.initialInterval = config.DateCursor.InitialInterval
 	c.dateFormat = config.DateCursor.getDateFormat()
-	c.valueTpl = config.DateCursor.ValueTemplate.Template
+	if config.DateCursor.ValueTemplate != nil {
+		c.valueTpl = config.DateCursor.ValueTemplate.Template
+	}
 
 	return c
 }
diff --git a/x-pack/filebeat/input/httpjson/pagination.go b/x-pack/filebeat/input/httpjson/pagination.go
@@ -72,7 +72,7 @@ func (p *pagination) nextRequestInfo(ri *requestInfo, response response, lastObj
 
 // getNextLinkFromHeader retrieves the next URL for pagination from the HTTP Header of the response
 func getNextLinkFromHeader(header http.Header, fieldName string, re *regexp.Regexp) (string, error) {
-	links, ok := header[fieldName]
+	links, ok := header[http.CanonicalHeaderKey(fieldName)]
 	if !ok {
 		return "", fmt.Errorf("field %s does not exist in the HTTP Header", fieldName)
 	}
diff --git a/x-pack/filebeat/input/httpjson/pagination_test.go b/x-pack/filebeat/input/httpjson/pagination_test.go
@@ -14,7 +14,7 @@ import (
 
 func TestGetNextLinkFromHeader(t *testing.T) {
 	header := make(http.Header)
-	header.Add("Link", "<https://dev-168980.okta.com/api/v1/logs>; rel=\"self\"")
+	header.Add("link", "<https://dev-168980.okta.com/api/v1/logs>; rel=\"self\"")
 	header.Add("Link", "<https://dev-168980.okta.com/api/v1/logs?after=1581658181086_1>; rel=\"next\"")
 	re, _ := regexp.Compile("<([^>]+)>; *rel=\"next\"(?:,|$)")
 	url, err := getNextLinkFromHeader(header, "Link", re)
diff --git a/x-pack/filebeat/input/httpjson/requester.go b/x-pack/filebeat/input/httpjson/requester.go
@@ -113,6 +113,7 @@ func (r *requester) processHTTPRequest(ctx context.Context, publisher cursor.Pub
 			return err
 		}
 
+		response.header = resp.Header
 		responseData, err := ioutil.ReadAll(resp.Body)
 		if err != nil {
 			return fmt.Errorf("failed to read http response: %w", err)
@@ -165,10 +166,10 @@ func (r *requester) processHTTPRequest(ctx context.Context, publisher cursor.Pub
 		if err != nil {
 			return err
 		}
-	}
 
-	if lastObj != nil && r.dateCursor.enabled {
-		r.updateCursorState(ri.url, r.dateCursor.getNextValue(common.MapStr(lastObj)))
+		if lastObj != nil && r.dateCursor.enabled {
+			r.updateCursorState(ri.url, r.dateCursor.getNextValue(common.MapStr(lastObj)))
+		}
 	}
 
 	return nil
diff --git a/x-pack/filebeat/module/okta/system/config/input.yml b/x-pack/filebeat/module/okta/system/config/input.yml
@@ -44,6 +44,10 @@ ssl: {{ .ssl | tojson }}
 url: {{ .url }}
 {{ end }}
 
+date_cursor.field: published
+date_cursor.url_field: since
+date_cursor.initial_interval: {{ .initial_interval }}
+
 {{ else if eq .input "file" }}
 
 type: log
diff --git a/x-pack/filebeat/module/okta/system/manifest.yml b/x-pack/filebeat/module/okta/system/manifest.yml
@@ -8,6 +8,7 @@ var:
     default: "SSWS"
   - name: http_client_timeout
   - name: http_method
+    default: GET
   - name: http_headers
   - name: http_request_body
   - name: interval
@@ -31,6 +32,8 @@ var:
   - name: tags
     default: [forwarded]
   - name: url
+  - name: initial_interval
+    default: 24h
 
 input: config/input.yml
 ingest_pipeline: ingest/pipeline.yml

Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ func (p pagination) nextRequestInfo(ri requestInfo, response response, lastObj`
`72`	`72`
`73`	`73`	`// getNextLinkFromHeader retrieves the next URL for pagination from the HTTP Header of the response`
`74`	`74`	`func getNextLinkFromHeader(header http.Header, fieldName string, re *regexp.Regexp) (string, error) {`
`75`		`- links, ok := header[fieldName]`
	`75`	`+ links, ok := header[http.CanonicalHeaderKey(fieldName)]`
`76`	`76`	`if !ok {`
`77`	`77`	`return "", fmt.Errorf("field %s does not exist in the HTTP Header", fieldName)`
`78`	`78`	`}`
Original file line number	Diff line number	Diff line change
`@@ -113,6 +113,7 @@ func (r *requester) processHTTPRequest(ctx context.Context, publisher cursor.Pub`
`113`	`113`	`return err`
`114`	`114`	`}`
`115`	`115`
	`116`	`+ response.header = resp.Header`
`116`	`117`	`responseData, err := ioutil.ReadAll(resp.Body)`
`117`	`118`	`if err != nil {`
`118`	`119`	`return fmt.Errorf("failed to read http response: %w", err)`
`@@ -165,10 +166,10 @@ func (r *requester) processHTTPRequest(ctx context.Context, publisher cursor.Pub`
`165`	`166`	`if err != nil {`
`166`	`167`	`return err`
`167`	`168`	`}`
`168`		`- }`
`169`	`169`
`170`		`- if lastObj != nil && r.dateCursor.enabled {`
`171`		`- r.updateCursorState(ri.url, r.dateCursor.getNextValue(common.MapStr(lastObj)))`
	`170`	`+ if lastObj != nil && r.dateCursor.enabled {`
	`171`	`+ r.updateCursorState(ri.url, r.dateCursor.getNextValue(common.MapStr(lastObj)))`
	`172`	`+ }`
`172`	`173`	`}`
`173`	`174`
`174`	`175`	`return nil`