Add Close() method to Project to release resources

Problem Observed

Goroutine leaks because there is no way to "close" projects

Problem Detail

NewProject() creates a defaultListener in it.
The listener starts a new goroutine in NewDefaultListener().
We currently don't have a method to stop this goroutine.

Suggested Resolution

This commit adds a new method Close() to release resources
tied to a Project.

Signed-off-by: Iwasaki Yudai <[email protected]>

Author: yudai

project/interface.go

@@ -39,6 +39,7 @@ type APIProject interface {
 	AddConfig(name string, config *config.ServiceConfig) error
 	Load(bytes []byte) error
 	Containers(ctx context.Context, filter Filter, services ...string) ([]string, error)
+	Close() error
 
 	GetServiceConfig(service string) (*config.ServiceConfig, bool)
 }

project/listener.go

@@ -32,24 +32,29 @@ var (
 	}
 )
 
-type defaultListener struct {
-	project    *Project
-	listenChan chan events.Event
-	upCount    int
+// DefaultListener is a listener with the default logger
+type DefaultListener struct {
+	// Event channel
+	C chan events.Event
+
+	project *Project
+	upCount int
 }
 
 // NewDefaultListener create a default listener for the specified project.
-func NewDefaultListener(p *Project) chan<- events.Event {
-	l := defaultListener{
-		listenChan: make(chan events.Event),
-		project:    p,
+func NewDefaultListener(p *Project) *DefaultListener {
+	l := &DefaultListener{
+		C:       make(chan events.Event),
+		project: p,
 	}
-	go l.start()
-	return l.listenChan
+	go l.Start()
+	return l
 }
 
-func (d *defaultListener) start() {
-	for event := range d.listenChan {
+// Start runs the main event loop.
+// This method blocks until Close() is called.
+func (d *DefaultListener) Start() {
+	for event := range d.C {
 		buffer := bytes.NewBuffer(nil)
 		if event.Data != nil {
 			for k, v := range event.Data {
@@ -79,3 +84,8 @@ func (d *defaultListener) start() {
 		}
 	}
 }
+
+// Close terminates event channel C.
+func (d *DefaultListener) Close() {
+	close(d.C)
+}

project/project.go

@@ -35,15 +35,15 @@ type Project struct {
 	ReloadCallback func() error
 	ParseOptions   *config.ParseOptions
 
-	runtime       RuntimeProject
-	networks      Networks
-	volumes       Volumes
-	configVersion string
-	context       *Context
-	reload        []string
-	upCount       int
-	listeners     []chan<- events.Event
-	hasListeners  bool
+	runtime         RuntimeProject
+	networks        Networks
+	volumes         Volumes
+	configVersion   string
+	context         *Context
+	reload          []string
+	upCount         int
+	listeners       []chan<- events.Event
+	defaultListener *DefaultListener
 }
 
 // NewProject creates a new project with the specified context.
@@ -93,11 +93,18 @@ func NewProject(context *Context, runtime RuntimeProject, parseOptions *config.P
 
 	context.Project = p
 
-	p.listeners = []chan<- events.Event{NewDefaultListener(p)}
+	p.defaultListener = NewDefaultListener(p)
+	p.listeners = []chan<- events.Event{p.defaultListener.C}
 
 	return p
 }
 
+// Close releases resources attached to the project
+func (p *Project) Close() error {
+	p.defaultListener.Close()
+	return nil
+}
+
 // Parse populates project information based on its context. It sets up the name,
 // the composefile and the composebytes (the composefile content).
 func (p *Project) Parse() error {
@@ -511,11 +518,7 @@ func (p *Project) traverse(start bool, selected map[string]bool, wrappers map[st
 // AddListener adds the specified listener to the project.
 // This implements implicitly events.Emitter.
 func (p *Project) AddListener(c chan<- events.Event) {
-	if !p.hasListeners {
-		for _, l := range p.listeners {
-			close(l)
-		}
-		p.hasListeners = true
+	if len(p.listeners) == 1 && p.listeners[0] == p.defaultListener.C {
 		p.listeners = []chan<- events.Event{c}
 	} else {
 		p.listeners = append(p.listeners, c)

project/project_test.go

@@ -64,6 +64,7 @@ func TestTwoCall(t *testing.T) {
 	p := NewProject(&Context{
 		ServiceFactory: factory,
 	}, nil, nil)
+	defer p.Close()
 	p.ServiceConfigs = config.NewServiceConfigs()
 	p.ServiceConfigs.Add("foo", &config.ServiceConfig{})
 
@@ -83,6 +84,7 @@ func TestTwoCall(t *testing.T) {
 func TestGetServiceConfig(t *testing.T) {
 
 	p := NewProject(&Context{}, nil, nil)
+	defer p.Close()
 	p.ServiceConfigs = config.NewServiceConfigs()
 	fooService := &config.ServiceConfig{}
 	p.ServiceConfigs.Add("foo", fooService)
@@ -112,6 +114,7 @@ func TestParseWithBadContent(t *testing.T) {
 			[]byte("garbage"),
 		},
 	}, nil, nil)
+	defer p.Close()
 
 	err := p.Parse()
 	if err == nil {
@@ -129,6 +132,7 @@ func TestParseWithGoodContent(t *testing.T) {
 			[]byte("not-garbage:\n  image: foo"),
 		},
 	}, nil, nil)
+	defer p.Close()
 
 	err := p.Parse()
 	if err != nil {
@@ -142,6 +146,7 @@ func TestParseWithDefaultEnvironmentLookup(t *testing.T) {
 			[]byte("not-garbage:\n  image: foo:${version}"),
 		},
 	}, nil, nil)
+	defer p.Close()
 
 	err := p.Parse()
 	if err != nil {
@@ -165,6 +170,7 @@ func TestEnvironmentResolve(t *testing.T) {
 		ServiceFactory:    factory,
 		EnvironmentLookup: &TestEnvironmentLookup{},
 	}, nil, nil)
+	defer p.Close()
 	p.ServiceConfigs = config.NewServiceConfigs()
 	p.ServiceConfigs.Add("foo", &config.ServiceConfig{
 		Environment: yaml.MaporEqualSlice([]string{
@@ -209,6 +215,7 @@ func TestParseWithMultipleComposeFiles(t *testing.T) {
 	p := NewProject(&Context{
 		ComposeBytes: [][]byte{configOne, configTwo},
 	}, nil, nil)
+	defer p.Close()
 
 	err := p.Parse()
 
@@ -222,6 +229,7 @@ func TestParseWithMultipleComposeFiles(t *testing.T) {
 	p = NewProject(&Context{
 		ComposeBytes: [][]byte{configTwo, configOne},
 	}, nil, nil)
+	defer p.Close()
 
 	err = p.Parse()
 
@@ -235,6 +243,7 @@ func TestParseWithMultipleComposeFiles(t *testing.T) {
 	p = NewProject(&Context{
 		ComposeBytes: [][]byte{configOne, configTwo, configThree},
 	}, nil, nil)
+	defer p.Close()
 
 	err = p.Parse()