From 0255c570710c7d1779ee5ab6cfdf2fd9d736c700 Mon Sep 17 00:00:00 2001
From: Chris Mark <chrismarkou92@gmail.com>
Date: Thu, 30 Jul 2020 12:25:08 +0300
Subject: [PATCH] Cherry-pick #20305 to 7.9: [Autodiscovery] Ignore
 ErrInputNotFinished errors in autodiscover config checks (#20338)

---
 CHANGELOG.next.asciidoc                       |  1 +
 filebeat/input/errors.go                      | 32 +++++++++++++++++++
 filebeat/input/file/state.go                  | 17 ++++++++++
 filebeat/input/log/input.go                   |  2 +-
 filebeat/input/runnerfactory.go               |  4 +++
 .../autodiscover/providers/kubernetes/pod.go  |  1 +
 6 files changed, 56 insertions(+), 1 deletion(-)
 create mode 100644 filebeat/input/errors.go

diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
index 20951d38bdd4..86cd022f1849 100644
--- a/CHANGELOG.next.asciidoc
+++ b/CHANGELOG.next.asciidoc
@@ -93,6 +93,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix terminating pod autodiscover issue. {pull}20084[20084]
 - Fix seccomp policy for calls to `chmod` and `chown`. {pull}20054[20054]
 - Output errors when Kibana index pattern setup fails. {pull}20121[20121]
+- Fix issue in autodiscover that kept inputs stopped after config updates. {pull}20305[20305]
 
 *Auditbeat*
 
diff --git a/filebeat/input/errors.go b/filebeat/input/errors.go
new file mode 100644
index 000000000000..098156abf91d
--- /dev/null
+++ b/filebeat/input/errors.go
@@ -0,0 +1,32 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package input
+
+import (
+	"fmt"
+)
+
+// ErrInputNotFinished struct for reporting errors related to not finished inputs
+type ErrInputNotFinished struct {
+	State string
+}
+
+// Error method of ErrInputNotFinished
+func (e *ErrInputNotFinished) Error() string {
+	return fmt.Sprintf("Can only start an input when all related states are finished: %+v", e.State)
+}
diff --git a/filebeat/input/file/state.go b/filebeat/input/file/state.go
index 18fcfee583b0..560daf8a7bc7 100644
--- a/filebeat/input/file/state.go
+++ b/filebeat/input/file/state.go
@@ -18,6 +18,7 @@
 package file
 
 import (
+	"fmt"
 	"os"
 	"time"
 
@@ -66,3 +67,19 @@ func NewState(fileInfo os.FileInfo, path string, t string, meta map[string]strin
 func (s *State) IsEqual(c *State) bool {
 	return s.Id == c.Id
 }
+
+// String returns string representation of the struct
+func (s *State) String() string {
+	return fmt.Sprintf(
+		"{Id: %v, Finished: %v, Fileinfo: %v, Source: %v, Offset: %v, Timestamp: %v, TTL: %v, Type: %v, Meta: %v, FileStateOS: %v}",
+		s.Id,
+		s.Finished,
+		s.Fileinfo,
+		s.Source,
+		s.Offset,
+		s.Timestamp,
+		s.TTL,
+		s.Type,
+		s.Meta,
+		s.FileStateOS)
+}
diff --git a/filebeat/input/log/input.go b/filebeat/input/log/input.go
index d7bd4f180170..a1837bbd4712 100644
--- a/filebeat/input/log/input.go
+++ b/filebeat/input/log/input.go
@@ -175,7 +175,7 @@ func (p *Input) loadStates(states []file.State) error {
 
 			// In case a input is tried to be started with an unfinished state matching the glob pattern
 			if !state.Finished {
-				return fmt.Errorf("Can only start an input when all related states are finished: %+v", state)
+				return &input.ErrInputNotFinished{State: state.String()}
 			}
 
 			// Convert state to current identifier if different
diff --git a/filebeat/input/runnerfactory.go b/filebeat/input/runnerfactory.go
index 179057c43737..afd67d4e08ab 100644
--- a/filebeat/input/runnerfactory.go
+++ b/filebeat/input/runnerfactory.go
@@ -59,5 +59,9 @@ func (r *RunnerFactory) Create(
 
 func (r *RunnerFactory) CheckConfig(cfg *common.Config) error {
 	_, err := r.Create(pipeline.NewNilPipeline(), cfg)
+	if _, ok := err.(*ErrInputNotFinished); ok {
+		// error is related to state, and hence config can be considered valid
+		return nil
+	}
 	return err
 }
diff --git a/libbeat/autodiscover/providers/kubernetes/pod.go b/libbeat/autodiscover/providers/kubernetes/pod.go
index 39df134b8097..033146a84d44 100644
--- a/libbeat/autodiscover/providers/kubernetes/pod.go
+++ b/libbeat/autodiscover/providers/kubernetes/pod.go
@@ -160,6 +160,7 @@ func (p *pod) OnUpdate(obj interface{}) {
 			}
 		}
 		time.AfterFunc(p.config.CleanupTimeout, func() { p.emit(pod, "stop") })
+		return
 	}
 
 	p.logger.Debugf("Watcher Pod update: %+v", obj)