Merge branch 'versions/0.10' into dev

Conflicts:
	cmd/hekad/main.go
	docs/source/conf.py

Author: rafrombrc

CHANGES.txt

@@ -17,6 +17,32 @@ Features
 
 * Added decoder support to StatAccumInput.
 
+* Added `git_clone_to_path` to the cmake build to allow git repos to be cloned
+  into alternate locations; useful for relocating forks of Go packages into
+  their original import paths.
+
+0.10.0 (2015-??-??)
+===================
+
+0.10.0b2 (2015-11-20)
+=====================
+
+Backwards Incompatibilities
+---------------------------
+
+* StatAccumInput Input: percent_threshold param type convert to slice
+
+Bug Handling
+------------
+
+* Updated Sarama dependency from pre-1.0 release fork to fork (with only test
+  code changes) of Sarama 1.5.0 release.
+
+Features
+--------
+
+* Added decoder support to StatAccumInput.
+
 * Added `git_clone_to_path` to the cmake build to allow git repos to be cloned
   into alternate locations; useful for relocating forks of Go packages into
   their original import paths.
@@ -49,6 +75,17 @@ Features
 Bug Handling
 ------------
 
+* Fixed issue where ElasticSearchOutput was trying to send zero length requests
+  (#1783).
+
+* Fixed race condition in ElasticSearchOutput (#1786).
+
+* AMQPInput `Run` method now returns an error when the input channel closes but
+  `Stop` hasn't been called to successfully trigger restarts (#1757).
+
+* Fixed error where restarting plugins were losing specified configuration
+  (#1756).
+
 * Fixed config error where global `max_pack_idle` setting was the wrong type
   and was being ignored (#1778).
 
@@ -87,6 +124,9 @@ Bug Handling
 * Fixed panic that was occurring when loading a config file or directory that
   exists but which registers no plugins (#1597).
 
+* Delay start up when a buffered plugin's buffer is at capacity to give the
+  back-pressure time to resolve (#1738).
+
 * Fixed bug where LogStreamerInput would sometimes loop infinitely reading the
   same file over and over when reading gzipped log files.
 

cmd/hekad/config.go

@@ -47,6 +47,7 @@ type HekadConfig struct {
 	Hostname              string
 	MaxMessageSize        uint32 `toml:"max_message_size"`
 	LogFlags              int    `toml:"log_flags"`
+	FullBufferMaxRetries  uint32 `toml:"full_buffer_max_retries"`
 }
 
 func LoadHekadConfig(configPath string) (config *HekadConfig, err error) {
@@ -71,6 +72,7 @@ func LoadHekadConfig(configPath string) (config *HekadConfig, err error) {
 		PidFile:               "",
 		Hostname:              hostname,
 		LogFlags:              log.LstdFlags,
+		FullBufferMaxRetries:  10,
 	}
 
 	var configFile map[string]toml.Primitive

cmd/hekad/main.go

@@ -87,6 +87,7 @@ func setGlobalConfigs(config *HekadConfig) (*pipeline.GlobalConfigStruct, string
 	globals.ShareDir = config.ShareDir
 	globals.SampleDenominator = config.SampleDenominator
 	globals.Hostname = config.Hostname
+	globals.FullBufferMaxRetries = uint(config.FullBufferMaxRetries)
 
 	return globals, cpuProfName, memProfName
 }

docs/source/config/index.rst

@@ -171,6 +171,14 @@ Config:
     and time, the default) or 0 (no prefix). See
     `https://golang.org/pkg/log/#pkg-constants Go documentation`_ for details.
 
+- full_buffer_max_retries (int):
+    When Heka shuts down due to a buffer filling to capacity, the next time
+    Heka starts it will delay startup briefly to give the buffer a chance to
+    drain, to alleviate the back-pressure. This setting specifies the maximum
+    number of intervals (max 1s in duration) Heka should wait for the buffer
+    size to get below 90% of capacity before deciding that the issue is not
+    resolved and continuing startup (or shutting down).
+
 Example hekad.toml file
 =======================
 

pipeline/pipeline_runner.go

@@ -51,12 +51,15 @@ type GlobalConfigStruct struct {
 	MaxPackIdle           time.Duration
 	stopping              bool
 	stoppingMutex         sync.RWMutex
+	shutdownOnce          sync.Once
 	BaseDir               string
 	ShareDir              string
 	SampleDenominator     int
 	sigChan               chan os.Signal
 	Hostname              string
 	abortChan             chan struct{}
+	FullBufferMaxRetries  uint
+	exitCode              int
 }
 
 // Creates a GlobalConfigStruct object populated w/ default values.
@@ -87,10 +90,13 @@ func (g *GlobalConfigStruct) SigChan() chan os.Signal {
 // This method returns immediately by spawning a goroutine to do to
 // work so that the caller won't end up blocking part of the shutdown
 // sequence
-func (g *GlobalConfigStruct) ShutDown() {
-	go func() {
-		g.sigChan <- syscall.SIGINT
-	}()
+func (g *GlobalConfigStruct) ShutDown(exitCode int) {
+	g.shutdownOnce.Do(func() {
+		g.exitCode = exitCode
+		go func() {
+			g.sigChan <- syscall.SIGINT
+		}()
+	})
 }
 
 func (g *GlobalConfigStruct) IsShuttingDown() (stopping bool) {
@@ -251,9 +257,9 @@ func (p *PipelinePack) EncodeMsgBytes() error {
 	return err
 }
 
-// Main function driving Heka execution. Loads config, initializes
-// PipelinePack pools, and starts all the runners. Then it listens for signals
-// and drives the shutdown process when that is triggered.
+// Main function driving Heka execution. Loads config, initializes PipelinePack
+// pools, and starts all the runners. Then it listens for signals and drives
+// the shutdown process when that is triggered.
 func Run(config *PipelineConfig) {
 	LogInfo.Println("Starting hekad...")
 
@@ -268,7 +274,7 @@ func Run(config *PipelineConfig) {
 			LogError.Printf("Output '%s' failed to start: %s", name, err)
 			outputsWg.Done()
 			if !output.IsStoppable() {
-				globals.ShutDown()
+				globals.ShutDown(1)
 			}
 			continue
 		}
@@ -281,7 +287,7 @@ func Run(config *PipelineConfig) {
 			LogError.Printf("Filter '%s' failed to start: %s", name, err)
 			config.filtersWg.Done()
 			if !filter.IsStoppable() {
-				globals.ShutDown()
+				globals.ShutDown(1)
 			}
 			continue
 		}
@@ -319,7 +325,7 @@ func Run(config *PipelineConfig) {
 			LogError.Printf("Input '%s' failed to start: %s", name, err)
 			config.inputsWg.Done()
 			if !input.IsStoppable() {
-				globals.ShutDown()
+				globals.ShutDown(1)
 			}
 			continue
 		}
@@ -398,6 +404,7 @@ func Run(config *PipelineConfig) {
 	}
 
 	LogInfo.Println("Shutdown complete.")
+	os.Exit(globals.exitCode)
 }
 
 func sandboxAbort(config *PipelineConfig) {
@@ -442,6 +449,6 @@ func sandboxAbort(config *PipelineConfig) {
 	// and abort any sandboxes that are wedged inside process_message or
 	// timer_event.
 	config.allReportsStdout()
-	config.Globals.ShutDown()
+	config.Globals.ShutDown(1)
 	close(config.Globals.abortChan)
 }

pipeline/plugin_runners.go

@@ -199,21 +199,21 @@ type InputRunner interface {
 
 type iRunner struct {
 	pRunnerBase
-	input               Input
-	config              CommonInputConfig
-	pConfig             *PipelineConfig
-	inChan              chan *PipelinePack
-	ticker              <-chan time.Time
-	transient           bool
-	syncDecode          bool
-	sendDecodeFailures  bool
-	logDecodeFailures   bool
-	deliver             DeliverFunc
-	delivererOnce       sync.Once
-	delivererLock       sync.Mutex
-	canExit             bool
-	shutdownWanters     []WantsDecoderRunnerShutdown
-	shutdownLock        sync.Mutex
+	input              Input
+	config             CommonInputConfig
+	pConfig            *PipelineConfig
+	inChan             chan *PipelinePack
+	ticker             <-chan time.Time
+	transient          bool
+	syncDecode         bool
+	sendDecodeFailures bool
+	logDecodeFailures  bool
+	deliver            DeliverFunc
+	delivererOnce      sync.Once
+	delivererLock      sync.Mutex
+	canExit            bool
+	shutdownWanters    []WantsDecoderRunnerShutdown
+	shutdownLock       sync.Mutex
 }
 
 func (ir *iRunner) Ticker() (ticker <-chan time.Time) {
@@ -314,7 +314,7 @@ func (ir *iRunner) Starter(h PluginHelper, wg *sync.WaitGroup) {
 	if err != nil {
 		ir.LogError(err)
 		if !ir.IsStoppable() {
-			globals.ShutDown()
+			globals.ShutDown(1)
 		}
 		return
 	}
@@ -344,7 +344,7 @@ func (ir *iRunner) Starter(h PluginHelper, wg *sync.WaitGroup) {
 				break
 			}
 
-			// Otherwise we'll execute the Retry config
+			// Otherwise we'll execute the Retry config.
 			recon.CleanupForRestart()
 			if ir.maker == nil {
 				ir.pConfig.makersLock.RLock()
@@ -364,10 +364,16 @@ func (ir *iRunner) Starter(h PluginHelper, wg *sync.WaitGroup) {
 			ir.LogMessage(fmt.Sprintf("Restarting (attempt %d/%d)\n",
 				rh.times, rh.retries))
 
-			// If we've not been created elsewhere, call the plugin's Init()
+			// If we've not been created elsewhere, call the plugin's Init().
 			if !ir.transient {
-				if err = ir.plugin.Init(ir.maker.Config()); err != nil {
-					// We couldn't reInit the plugin, do a mini-retry loop
+				var config interface{}
+				if config, err = ir.maker.PrepConfig(); err != nil {
+					// We couldn't reInit the plugin, do a mini-retry loop.
+					ir.LogError(err)
+					goto initLoop
+				}
+				if err = ir.plugin.Init(config); err != nil {
+					// We couldn't reInit the plugin, do a mini-retry loop.
 					ir.LogError(err)
 					goto initLoop
 				}
@@ -380,7 +386,7 @@ func (ir *iRunner) Starter(h PluginHelper, wg *sync.WaitGroup) {
 
 	// If we're not a stoppable input, trigger Heka shutdown.
 	if !ir.IsStoppable() {
-		globals.ShutDown()
+		globals.ShutDown(1)
 	}
 }
 
@@ -957,10 +963,43 @@ func (foRunner *foRunner) BackPressured() bool {
 		return len(foRunner.inChan) >= foRunner.capacity ||
 			foRunner.matcher.InChanLen() >= foRunner.capacity
 	}
-
 	return foRunner.capacity > 0 && foRunner.bufReader.queueSize.Get() >= uint64(foRunner.capacity)
 }
 
+func (foRunner *foRunner) waitForBackPressure() error {
+	globals := foRunner.pConfig.Globals
+	retryOptions := getDefaultRetryOptions()
+	retryOptions.MaxDelay = "1s"
+	retryOptions.MaxRetries = int(globals.FullBufferMaxRetries)
+	// NewRetryHelper will only return an error if the duration strings don't
+	// parse. Ours are hard-coded, so this error shouldn't happen.
+	retry, err := NewRetryHelper(retryOptions)
+	if err != nil {
+		return fmt.Errorf("can't create retry helper: %s", err.Error())
+	}
+	for !globals.IsShuttingDown() {
+		bp := foRunner.BackPressured()
+		fmt.Println("back-pressured?: ", bp)
+		if !bp {
+			return nil
+		}
+		err = retry.Wait()
+		if err != nil {
+			// We've exhausted our max allowed retries, so we honor the
+			// buffer's 'full_action' setting and trigger a shutdown if
+			// necessary.
+			if foRunner.bufReader.config.FullAction == "shutdown" {
+				globals.ShutDown(1)
+				foRunner.LogError(errors.New("back-pressure not resolving: triggering shutdown"))
+			}
+			// But we always return `nil` so that regular start up sequence can
+			// continue.
+			return nil
+		}
+	}
+	return nil
+}
+
 func (foRunner *foRunner) Start(h PluginHelper, wg *sync.WaitGroup) (err error) {
 	foRunner.h = h
 	foRunner.pConfig = h.PipelineConfig()
@@ -1034,6 +1073,14 @@ func (foRunner *foRunner) Start(h PluginHelper, wg *sync.WaitGroup) (err error)
 	} else {
 		go foRunner.OldStarter(h, wg)
 	}
+
+	if foRunner.useBuffering && foRunner.BackPressured() {
+		foRunner.LogMessage("Delaying start while trying to relieve back-pressure...")
+		if err = foRunner.waitForBackPressure(); err != nil {
+			return err
+		}
+	}
+
 	return
 }
 
@@ -1145,7 +1192,7 @@ func (foRunner *foRunner) Starter(plugin MessageProcessor, h PluginHelper,
 	if err != nil {
 		foRunner.LogError(err)
 		if !foRunner.IsStoppable() {
-			globals.ShutDown()
+			globals.ShutDown(1)
 		}
 		return
 	}
@@ -1183,7 +1230,7 @@ func (foRunner *foRunner) Starter(plugin MessageProcessor, h PluginHelper,
 			// No more retries.
 			foRunner.lastErr = err
 			if !foRunner.IsStoppable() {
-				globals.ShutDown()
+				globals.ShutDown(1)
 			}
 			return
 		}
@@ -1251,7 +1298,12 @@ func (foRunner *foRunner) Starter(plugin MessageProcessor, h PluginHelper,
 			break
 		}
 		foRunner.LogMessage("now restarting")
-		if err = foRunner.plugin.Init(foRunner.maker.Config()); err != nil {
+		var config interface{}
+		if config, err = foRunner.maker.PrepConfig(); err != nil {
+			foRunner.LogError(err)
+			goto initLoop
+		}
+		if err = foRunner.plugin.Init(config); err != nil {
 			foRunner.LogError(err)
 			goto initLoop
 		}
@@ -1308,7 +1360,7 @@ func (foRunner *foRunner) exit() {
 	// Also, if this isn't a "stoppable" plugin we shut everything down.
 	if !foRunner.IsStoppable() {
 		foRunner.LogMessage("has stopped, shutting down.")
-		foRunner.pConfig.Globals.ShutDown()
+		foRunner.pConfig.Globals.ShutDown(1)
 		return
 	}
 
@@ -1424,7 +1476,7 @@ func (foRunner *foRunner) OldStarter(helper PluginHelper, wg *sync.WaitGroup) {
 	if err != nil {
 		foRunner.LogError(err)
 		if !foRunner.IsStoppable() {
-			globals.ShutDown()
+			globals.ShutDown(1)
 		}
 		return
 	}
@@ -1497,7 +1549,12 @@ func (foRunner *foRunner) OldStarter(helper PluginHelper, wg *sync.WaitGroup) {
 			break
 		}
 		foRunner.LogMessage("now restarting")
-		if err = foRunner.plugin.Init(foRunner.maker.Config()); err != nil {
+		var config interface{}
+		if config, err = foRunner.maker.PrepConfig(); err != nil {
+			foRunner.LogError(err)
+			goto initLoop
+		}
+		if err = foRunner.plugin.Init(config); err != nil {
 			foRunner.LogError(err)
 			goto initLoop
 		}