Merge pull request #70 from rchicoli/free-up-resource

Free up resource

Author: rchicoli

internal/pkg/errgroup/errgroup.go

@@ -0,0 +1,69 @@
+package errgroup
+
+import (
+	"sync"
+
+	"golang.org/x/net/context"
+)
+
+// A Group is a collection of goroutines working on subtasks that are part of
+// the same overall task.
+//
+// A zero Group is valid and does not cancel on error.
+type Group struct {
+	cancel func()
+
+	wg sync.WaitGroup
+
+	errOnce sync.Once
+	err     error
+}
+
+// WithContext returns a new Group and an associated Context derived from ctx.
+//
+// The derived Context is canceled the first time a function passed to Go
+// returns a non-nil error or the first time Wait returns, whichever occurs
+// first.
+func WithContext(ctx context.Context) (*Group, context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+	return &Group{cancel: cancel}, ctx
+}
+
+// Wait blocks until all function calls from the Go method have returned, then
+// returns the first non-nil error (if any) from them.
+func (g *Group) Wait() error {
+	g.wg.Wait()
+	if g.cancel != nil {
+		g.cancel()
+	}
+	return g.err
+}
+
+// Stop cancels all goroutines
+func (g *Group) Stop() error {
+	if g.cancel != nil {
+		g.cancel()
+	}
+	return g.err
+}
+
+// Go calls the given function in a new goroutine.
+//
+// The first call to return a non-nil error cancels the group; its error will be
+// returned by Wait.
+func (g *Group) Go(f func() error) {
+	g.wg.Add(1)
+
+	go func() {
+		defer g.wg.Done()
+
+		if err := f(); err != nil {
+			g.errOnce.Do(func() {
+				g.err = err
+				if g.cancel != nil {
+					g.cancel()
+				}
+			})
+		}
+	}()
+}

pkg/docker/driver.go

@@ -12,12 +12,11 @@ import (
 	"syscall"
 	"time"
 
-	"golang.org/x/sync/errgroup"
-
 	"github.com/docker/docker/api/types/plugins/logdriver"
 	"github.com/docker/docker/daemon/logger"
 	"github.com/tonistiigi/fifo"
 
+	"github.com/rchicoli/docker-log-elasticsearch/internal/pkg/errgroup"
 	"github.com/rchicoli/docker-log-elasticsearch/pkg/elasticsearch"
 	"github.com/rchicoli/docker-log-elasticsearch/pkg/extension/grok"
 
@@ -307,19 +306,12 @@ func (d *Driver) StopLogging(file string) error {
 		}
 	}
 
-	if d.esClient != nil {
-		// l.Printf("INFO client: %v", d.esClient)
-		if err := d.esClient.Close(); err != nil {
-			l.Printf("error: closing client connection: %v", err)
-		}
-		d.esClient.Stop()
-	}
-
 	if d.pipeline.group != nil {
 		// l.Printf("INFO with pipeline: %v", d.pipeline)
 
 		// close(d.pipeline.inputCh)
 		// close(d.pipeline.outputCh)
+		d.pipeline.group.Stop()
 		// d.pipeline.stopCh <- struct{}{}
 
 		// TODO: close channels gracefully
@@ -330,6 +322,15 @@ func (d *Driver) StopLogging(file string) error {
 		// 	l.Printf("error with pipeline: %v", err)
 		// }
 	}
+
+	if d.esClient != nil {
+		// l.Printf("INFO client: %v", d.esClient)
+		if err := d.esClient.Close(); err != nil {
+			l.Printf("error: closing client connection: %v", err)
+		}
+		d.esClient.Stop()
+	}
+
 	// l.Printf("INFO done stop logging")
 
 	return nil