Merge pull request #65 from rchicoli/development

new release: elasticsearch bulk

Author: rchicoli

.travis.yml

@@ -57,6 +57,9 @@ matrix:
       if: type IN (pull_request)
 
     - stage: release
+      # env: MAJOR_RELEASE=true
+      # it would be nice to map github labels instead
+      env: RELEASE_TAG=0.3.0
       if: type IN (push)
       script:
         - ./scripts/git-release.sh

Dockerfile

@@ -1,4 +1,4 @@
-FROM  golang:1.9.2-alpine as builder
+FROM  golang:1.10-alpine as builder
 
 ARG GOOS=linux
 ARG GOARCH=amd64

Makefile

@@ -137,4 +137,8 @@ acceptance_tests:
 	bats $(TESTS_DIR)/acceptance-tests/$(BATS_TESTFILE)
 
 integration_tests:
-	bats $(TESTS_DIR)/integration-tests/$(BATS_TESTFILE)
+	bats $(TESTS_DIR)/integration-tests/$(BATS_TESTFILE)
+
+create_environment: deploy_elasticsearch deploy_webapper
+
+delete_environment: undeploy_elasticsearch undeploy_webapper

README.md

@@ -1,6 +1,6 @@
 # Docker Log Elasticsearch
 
-[![GoDoc][1]][2] [![License: Apache 2.0][3]][4] [![Release][5]][6] [![Travis Status][7]][8] [![Coveralls Coverage][9]][10] [![Go Report Card][13]][14] [![Downloads][15]][16]
+[![GoDoc][1]][2] [![License: Apache 2.0][3]][4] [![Release][5]][6] [![Travis Status][7]][8] [![Coveralls Coverage][9]][10] [![Go Report Card][11]][12]
 
 [1]: https://godoc.org/github.com/rchicoli/docker-log-elasticsearch?status.svg
 [2]: https://godoc.org/github.com/rchicoli/docker-log-elasticsearch
@@ -12,14 +12,9 @@
 [8]: https://travis-ci.org/rchicoli/docker-log-elasticsearch
 [9]: https://coveralls.io/repos/rchicoli/docker-log-elasticsearch/badge.svg?branch=master&service=github
 [10]: https://coveralls.io/github/rchicoli/docker-log-elasticsearch?branch=master
-[11]: https://codecov.io/github/rchicoli/docker-log-elasticsearch/coverage.svg?branch=master
-[12]: https://codecov.io/github/rchicoli/docker-log-elasticsearch?branch=master
-[13]: https://goreportcard.com/badge/github.com/rchicoli/docker-log-elasticsearch
-[14]: https://goreportcard.com/report/github.com/rchicoli/docker-log-elasticsearch
-[15]: https://img.shields.io/github/downloads/rchicoli/docker-log-elasticsearch/total.svg?maxAge=1800
-[16]: https://github.com/rchicoli/docker-log-elasticsearch/releases
-[17]: https://img.shields.io/github/stars/rchicoli/docker-log-elasticsearch.svg
-[18]: https://github.com/rchicoli/docker-log-elasticsearch/stargazers
+[11]: https://goreportcard.com/badge/github.com/rchicoli/docker-log-elasticsearch
+[12]: https://goreportcard.com/report/github.com/rchicoli/docker-log-elasticsearch
+
 
 `docker-log-elasticsearch` forwards container logs to Elasticsearch service.
 
@@ -69,6 +64,10 @@ Before creating a docker container, a healthy instance of Elasticsearch service
 | elasticsearch-username | no | no |  |
 | elasticsearch-url   | no     | yes |
 | elasticsearch-version | 5 | no |
+| elasticsearch-bulk-actions | 100 | no |
+| elasticsearch-bulk-size | 5242880 | no |
+| elasticsearch-bulk-flush-interval | 5s | no |
+| elasticsearch-bulk-workers | 1 | no |
 | grok-named-capture | true | no |
 | grok-pattern | no | no |
 | grok-pattern-from | no | no |
@@ -119,6 +118,22 @@ Before creating a docker container, a healthy instance of Elasticsearch service
   - *version* of Elasticsearch cluster
   - *examples*: 1, 2, 5, 6
 
+###### elasticsearch-bulk-actions ######
+  - *bulk-actions* specifies when to flush based on the number of actions currently added
+  - *examples*: 1000
+
+###### elasticsearch-bulk-workers ######
+  - *bulk-workers* is the number of concurrent workers allowed to be executed
+  - *examples*: 2
+
+###### elasticsearch-bulk-size ######
+  - *bulk-size* specifies when to flush based on the size (in bytes) of the actions currently added. Defaults to 5 MB and can be set to -1 to be disabled.
+  - *examples*: 1024, -1
+
+###### elasticsearch-bulk-flush-interval ######
+  - *bulk-flush-interval* specifies when to flush at the end of the given interval
+  - *examples*: 300ms, 1s, 2h45m
+
 ###### grok-pattern ######
   - *pattern* add customer pattern
   - *examples*: CUSTOM_IP=(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)

ROADMAP.md

@@ -23,10 +23,11 @@ Goals:
  - [ ] Add the capability of multilines for Java Exceptions
  - [X] Add HTTPS Support and Skip Certificate Verify
  - [ ] Buffer logs into a file, if elasticsearch crashes. Add buffer size as well.
- - [ ] Implement bulk inserts
- - [ ] Implement queue size and batch size
    - [ ] if queue is full, then write to file
    - [ ] if file buffer is full, discard messages
+ - [X] Implement bulk inserts
+ - [X] Implement queue size and batch size
+   - [X] if number of requests or batch size have been reached, send messages to elasticsearch
  - [ ] Implement Readlog capability
  - [X] Implement grok for parsing docker logs
  - [ ] Add CONTRIBUTING file

pkg/docker/config.go

@@ -6,6 +6,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/docker/docker/daemon/logger"
 )
@@ -23,9 +24,20 @@ type LogOpt struct {
 	sniff    bool
 	insecure bool
 
+	Bulk
+
 	Grok
 }
 
+// Bulk configures the Bulk Processor Service
+type Bulk struct {
+	workers       int
+	actions       int
+	size          int
+	flushInterval time.Duration
+	stats         bool
+}
+
 // Grok ...
 type Grok struct {
 	grokPattern         string
@@ -46,6 +58,14 @@ func defaultLogOpt() *LogOpt {
 		sniff:    true,
 		insecure: false,
 
+		Bulk: Bulk{
+			workers:       1,
+			actions:       100,
+			size:          5 << 20,
+			flushInterval: 5 * time.Second,
+			stats:         false,
+		},
+
 		Grok: Grok{
 			grokPatternSplitter: " and ",
 			grokNamedCapture:    true,
@@ -139,6 +159,38 @@ func (c *LogOpt) validateLogOpt(cfg map[string]string) error {
 				return fmt.Errorf("error: parsing elasticsearch-timeout: %q", err)
 			}
 			c.timeout = timeout
+
+		case "elasticsearch-bulk-workers":
+			workers, err := strconv.Atoi(v)
+			if err != nil {
+				return fmt.Errorf("error: parsing elasticsearch-bulk-workers: %q", err)
+			}
+			c.Bulk.workers = workers
+		case "elasticsearch-bulk-actions":
+			actions, err := strconv.Atoi(v)
+			if err != nil {
+				return fmt.Errorf("error: parsing elasticsearch-bulk-actions: %q", err)
+			}
+			c.Bulk.actions = actions
+		case "elasticsearch-bulk-size":
+			size, err := strconv.Atoi(v)
+			if err != nil {
+				return fmt.Errorf("error: parsing elasticsearch-bulk-size: %q", err)
+			}
+			c.Bulk.size = size
+		case "elasticsearch-bulk-flush-interval":
+			flushInterval, err := time.ParseDuration(v)
+			if err != nil {
+				return fmt.Errorf("error: parsing elasticsearch-bulk-flush-interval: %q", err)
+			}
+			c.Bulk.flushInterval = flushInterval
+		case "elasticsearch-bulk-stats":
+			stats, err := strconv.ParseBool(v)
+			if err != nil {
+				return fmt.Errorf("error: parsing elasticsearch-bulk-stats: %q", err)
+			}
+			c.Bulk.stats = stats
+
 		case "grok-pattern":
 			c.grokPattern = v
 		case "grok-pattern-from":
@@ -153,7 +205,8 @@ func (c *LogOpt) validateLogOpt(cfg map[string]string) error {
 				return fmt.Errorf("error: parsing grok-named-capture: %q", err)
 			}
 			c.grokNamedCapture = s
-		// case "tag":
+
+			// case "tag":
 		// case "labels":
 		// case "env":
 		default: