forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
the FIELDS updated Notes field when parsing of a Table fails, when no exchange, the field exchange is no longer displayed instead of none, commented the two Timestamp tests that made the Travis build failed gofmted whole packetbeat code empty exchange not displayed, connection.close and channel.close are now waiting for a close ok method. if the close ok method never arrives, the transaction is published in the expireTransaction function and the Notes field is updated to indicate it Create a proper ICMP only BPF filter Update to golang 1.5.3 Filebeat system tests adapted to new structure Topbeat system tests updated Move implementation of packetbeat system tests to new base Not all code was moved to far, as there are some special implementations in packetbeat Update winlogbeat to the newest beat system tests Apply flake8 Fix 769: building the test binary includes all vendor package "go test -c -covermode=atomic -coverpkg" is called with a list of packages isntead of "./..." Add PyYAML to Windows vagrant box. Change system test kill_and_wait method to gracefully stop processes on Windows Fixes elastic#599 Refactor beat exit * Introduce Signal function which is called if using CTRL-C or similar * Run now returns an error and doesn't exist itself anymore * Fix spooler and crawler shutdown issue * Update mockbeat to check Run return error. Thanks to @cyrilleverrier for his contribution here. Fix elastic#779: libbeat/Makefile filters vendor folder all the files inside the ./vendor/* folder are now excluded when executing: make fmt make simplify make vet Edit new/changed content added to topbeat for 1.1 Minor fix to awkward sentence move preprocessor from libbeat and move packetbeat - remove preprocessor worker completely from libbeat - introduce transaction publisher in packetbeat to event processing: - GeoIP - normalize addresses - simplify sync/async publisher client - update new_protocol docs and changelog Validate length in parser Check length in pgsql parser before parsing column content in case column length > buffer length. changed results member of amqp struct to publish.Transactions, updated test file handling of a message splitted in several body frames, need to test it. began to parse connection methods handling of connection information methods, like connection.start, connection.open or channel.open, added a hide_connection_information in config set to true by default to choose to display them or not fixed bytes in bytes out for classic methods, added basic qos method in connection information option, channel and connections methods with an error code above or equal to 300 are always published, some new functions are added to see if method reflects an error in the protocol removed useless condition in expireTransaction added some pcap tests added three tests for amqp and their pcap files, corrected the fields.yml and added the amqp part in the packetbeat.yml.j2 for tests camelcased structs, created parser file updated the yaml with the last option, removed amqp.response field in the connectionStartOkMethod since it can hold the credentials of the client when a plain auth is selected fixed yaml, camelcased const, some style fixes in code added hasProperty function, added verbose in testing
- Loading branch information
Developpement Web Thomas Paulmyer
authored and
Thomas Paulmyer
committed
Feb 9, 2016
1 parent
31c7e04
commit e973db0
Showing
28 changed files
with
3,935 additions
and
5 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| FROM golang:1.5.3 | ||
| MAINTAINER Nicolas Ruflin <ruflin@elastic.co> | ||
|
|
||
| RUN set -x && \ | ||
| apt-get update && \ | ||
| apt-get install -y netcat && \ | ||
| apt-get clean | ||
|
|
||
|
|
||
| ## Install go package dependencies | ||
| RUN set -x \ | ||
| go get \ | ||
| github.com/pierrre/gotestcover \ | ||
| github.com/tsg/goautotest \ | ||
| golang.org/x/tools/cmd/cover \ | ||
| golang.org/x/tools/cmd/vet | ||
|
|
||
| COPY libbeat/scripts/docker-entrypoint.sh /entrypoint.sh | ||
|
|
||
| ENV GO15VENDOREXPERIMENT=1 | ||
|
|
||
| RUN mkdir -p /etc/pki/tls/certs | ||
| COPY testing/environments/docker/logstash/pki/tls/certs/logstash.crt /etc/pki/tls/certs/logstash.crt | ||
|
|
||
| # Create a copy of the respository inside the container. | ||
| COPY . /go/src/github.com/elastic/beats/ |
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,129 @@ | ||
| package beat | ||
|
|
||
| import ( | ||
| "time" | ||
|
|
||
| cfg "github.com/elastic/beats/filebeat/config" | ||
| "github.com/elastic/beats/filebeat/input" | ||
| "github.com/elastic/beats/libbeat/logp" | ||
| ) | ||
|
|
||
| type Spooler struct { | ||
| Filebeat *Filebeat | ||
| exit chan struct{} | ||
| nextFlushTime time.Time | ||
| spool []*input.FileEvent | ||
| Channel chan *input.FileEvent | ||
| } | ||
|
|
||
| func NewSpooler(filebeat *Filebeat) *Spooler { | ||
| spooler := &Spooler{ | ||
| Filebeat: filebeat, | ||
| exit: make(chan struct{}), | ||
| } | ||
|
|
||
| config := &spooler.Filebeat.FbConfig.Filebeat | ||
|
|
||
| // Set the next flush time | ||
| spooler.nextFlushTime = time.Now().Add(config.IdleTimeoutDuration) | ||
| spooler.Channel = make(chan *input.FileEvent, 16) | ||
|
|
||
| return spooler | ||
| } | ||
|
|
||
| func (spooler *Spooler) Config() error { | ||
| config := &spooler.Filebeat.FbConfig.Filebeat | ||
|
|
||
| // Set default pool size if value not set | ||
| if config.SpoolSize == 0 { | ||
| config.SpoolSize = cfg.DefaultSpoolSize | ||
| } | ||
|
|
||
| // Set default idle timeout if not set | ||
| if config.IdleTimeout == "" { | ||
| logp.Debug("spooler", "Set idleTimeoutDuration to %s", cfg.DefaultIdleTimeout) | ||
| // Set it to default | ||
| config.IdleTimeoutDuration = cfg.DefaultIdleTimeout | ||
| } else { | ||
| var err error | ||
|
|
||
| config.IdleTimeoutDuration, err = time.ParseDuration(config.IdleTimeout) | ||
|
|
||
| if err != nil { | ||
| logp.Warn("Failed to parse idle timeout duration '%s'. Error was: %v", config.IdleTimeout, err) | ||
| return err | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // Run runs the spooler | ||
| // It heartbeats periodically. If the last flush was longer than | ||
| // 'IdleTimeoutDuration' time ago, then we'll force a flush to prevent us from | ||
| // holding on to spooled events for too long. | ||
| func (s *Spooler) Run() { | ||
|
|
||
| config := &s.Filebeat.FbConfig.Filebeat | ||
|
|
||
| // Sets up ticket channel | ||
| ticker := time.NewTicker(config.IdleTimeoutDuration / 2) | ||
|
|
||
| s.spool = make([]*input.FileEvent, 0, config.SpoolSize) | ||
|
|
||
| logp.Info("Starting spooler: spool_size: %v; idle_timeout: %s", config.SpoolSize, config.IdleTimeoutDuration) | ||
|
|
||
| // Loops until running is set to false | ||
| for { | ||
| select { | ||
|
|
||
| case <-s.exit: | ||
| break | ||
| case event, ok := <-s.Channel: | ||
| if ok { | ||
| s.spool = append(s.spool, event) | ||
|
|
||
| // Spooler is full -> flush | ||
| if len(s.spool) == cap(s.spool) { | ||
| logp.Debug("spooler", "Flushing spooler because spooler full. Events flushed: %v", len(s.spool)) | ||
| s.flush() | ||
| } | ||
| } | ||
| case <-ticker.C: | ||
| // Flush periodically | ||
| if time.Now().After(s.nextFlushTime) { | ||
| logp.Debug("spooler", "Flushing spooler because of timeout. Events flushed: %v", len(s.spool)) | ||
| s.flush() | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Stop stops the spooler. Flushes events before stopping | ||
| func (s *Spooler) Stop() { | ||
| logp.Info("Stopping spooler") | ||
| close(s.exit) | ||
|
|
||
| // Flush again before exiting spooler and closes channel | ||
| logp.Info("Spooler last flush spooler") | ||
| s.flush() | ||
| close(s.Channel) | ||
| } | ||
|
|
||
| // flush flushes all event and sends them to the publisher | ||
| func (s *Spooler) flush() { | ||
| // Checks if any new objects | ||
| if len(s.spool) > 0 { | ||
|
|
||
| // copy buffer | ||
| tmpCopy := make([]*input.FileEvent, len(s.spool)) | ||
| copy(tmpCopy, s.spool) | ||
|
|
||
| // clear buffer | ||
| s.spool = s.spool[:0] | ||
|
|
||
| // send | ||
| s.Filebeat.publisherChan <- tmpCopy | ||
| } | ||
| s.nextFlushTime = time.Now().Add(s.Filebeat.FbConfig.Filebeat.IdleTimeoutDuration) | ||
| } |
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
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
Oops, something went wrong.