Fix IPC issues in Darwin

cr: https://code.amazon.com/reviews/CR-100210237

Author: VishnuKarthikRavindran

agent/ipc/messagebus/respondent.go

@@ -76,14 +76,14 @@ func (bus *MessageBus) ProcessHealthRequest() {
 	var msg []byte
 
 	defer func() {
-		if bus.healthChannel.IsConnect() {
+		if bus.healthChannel.IsChannelInitialized() {
 			if err = bus.healthChannel.Close(); err != nil {
 				bus.context.Log().Errorf("failed to close health channel: %v", err)
 			}
 		}
 	}()
 
-	for !bus.healthChannel.IsConnect() {
+	for !bus.healthChannel.IsDialSuccessful() {
 		if err = bus.dialToCoreAgentChannel(message.GetWorkerHealthRequest, message.GetWorkerHealthChannel); err != nil {
 			// This happens when worker started before core agent is
 			//   and when the amazon-ssm-agent is terminated milliseconds after the ssm-agent-worker has been started
@@ -140,14 +140,14 @@ func (bus *MessageBus) ProcessTerminationRequest() {
 	var err error
 	var msg []byte
 	defer func() {
-		if bus.terminationChannel.IsConnect() {
+		if bus.terminationChannel.IsChannelInitialized() {
 			if err = bus.terminationChannel.Close(); err != nil {
 				bus.context.Log().Errorf("failed to close termination channel: %v", err)
 			}
 		}
 	}()
 
-	for !bus.terminationChannel.IsConnect() {
+	for !bus.terminationChannel.IsDialSuccessful() {
 		if err = bus.dialToCoreAgentChannel(message.TerminateWorkerRequest, message.TerminationWorkerChannel); err != nil {
 			// This happens when worker started before core agent is
 			//   and when the amazon-ssm-agent is terminated milliseconds after the ssm-agent-worker has been started

agent/ipc/messagebus/respondent_test.go

@@ -72,7 +72,8 @@ func TestMessageBusTestSuite(t *testing.T) {
 }
 
 func (suite *MessageBusTestSuite) TestProcessTerminationRequest_Successful() {
-	suite.mockTerminateChannel.On("IsConnect").Return(true).Twice()
+	suite.mockTerminateChannel.On("IsChannelInitialized").Return(true).Once()
+	suite.mockTerminateChannel.On("IsDialSuccessful").Return(true).Once()
 	suite.mockTerminateChannel.On("Close").Return(nil).Once()
 
 	request := message.CreateTerminateWorkerRequest()
@@ -91,35 +92,36 @@ func (suite *MessageBusTestSuite) TestProcessTerminationRequest_Successful() {
 
 func (suite *MessageBusTestSuite) TestProcessTerminationRequest_SuccessfulConnectionRetry() {
 	// First try channel not connected but fails initialize
-	suite.mockTerminateChannel.On("IsConnect").Return(false).Once()
+	suite.mockTerminateChannel.On("IsDialSuccessful").Return(false).Once()
 	suite.mockTerminateChannel.On("Initialize", mock.Anything).Return(fmt.Errorf("SomeErr")).Once()
 	suite.mockTerminateChannel.On("Close").Return(nil).Once()
 
 	// Second try channel not connected but fails dial
-	suite.mockTerminateChannel.On("IsConnect").Return(false).Once()
-	suite.mockTerminateChannel.On("Initialize", mock.Anything).Return(nil).Once()
+	suite.mockTerminateChannel.On("IsDialSuccessful").Return(false).Once()
+	suite.mockTerminateChannel.On("Initialize", mock.Anything).Return(nil)
 	suite.mockTerminateChannel.On("Dial", mock.Anything).Return(fmt.Errorf("SomeDialError")).Once()
 	suite.mockTerminateChannel.On("Close").Return(nil).Once()
 
 	// Third try channel not connected but finally succeeds
-	suite.mockTerminateChannel.On("IsConnect").Return(false).Once()
-	suite.mockTerminateChannel.On("Initialize", mock.Anything).Return(nil).Once()
+	suite.mockTerminateChannel.On("IsDialSuccessful").Return(false).Once()
+	suite.mockTerminateChannel.On("Initialize", mock.Anything).Return(nil)
 	suite.mockTerminateChannel.On("Dial", mock.Anything).Return(nil).Once()
-
-	// Fourth call to isConnect succeeds, fourth call is for defer where it will call close
-	suite.mockTerminateChannel.On("IsConnect").Return(true).Twice()
-	suite.mockTerminateChannel.On("Close").Return(nil).Once()
+	suite.mockTerminateChannel.On("IsDialSuccessful").Return(true).Once()
 
 	request := message.CreateTerminateWorkerRequest()
 	requestString, _ := jsonutil.Marshal(request)
 	suite.mockTerminateChannel.On("Recv").Return([]byte(requestString), nil)
 	suite.mockTerminateChannel.On("Send", mock.Anything).Return(nil)
 
-	suite.messageBus.ProcessTerminationRequest()
+	// Fourth call to isConnect succeeds, fourth call is for defer where it will call close
+	suite.mockTerminateChannel.On("IsChannelInitialized").Return(true).Once()
+	suite.mockTerminateChannel.On("Close").Return(nil).Once()
 
-	suite.mockTerminateChannel.AssertExpectations(suite.T())
+	suite.messageBus.ProcessTerminationRequest()
 
 	// Assert termination channel connected and that a termination message is sent
 	suite.Assertions.Equal(true, <-suite.messageBus.GetTerminationChannelConnectedChan())
 	suite.Assertions.Equal(true, <-suite.messageBus.GetTerminationRequestChan())
+
+	suite.mockTerminateChannel.AssertExpectations(suite.T())
 }

common/channel/channel.go

@@ -15,6 +15,7 @@
 package channel
 
 import (
+	"errors"
 	"runtime"
 	"runtime/debug"
 	"time"
@@ -35,12 +36,17 @@ type IChannel interface {
 	SetOption(name string, value interface{}) error
 	Listen(addr string) error
 	Dial(addr string) error
-	IsConnect() bool
+	IsChannelInitialized() bool
+	IsDialSuccessful() bool
+	IsListenSuccessful() bool
 }
 
 var (
 	newNamedPipeChannelRef     = NewNamedPipeChannel
 	isDefaultChannelPresentRef = utils.IsDefaultChannelPresent
+
+	ErrIPCChannelClosed       = errors.New("channel is closed")
+	ErrDialListenUnSuccessful = errors.New("IPC connection is not established")
 )
 
 // GetChannelCreator returns function reference for channel creation based
@@ -55,11 +61,11 @@ func GetChannelCreator(log log.T, appConfig appconfig.SsmagentConfig, identity i
 // canUseNamedPipe checks whether named pipe can be used for IPC or not
 func canUseNamedPipe(log log.T, appConfig appconfig.SsmagentConfig, identity identity.IAgentIdentity) (useNamedPipe bool) {
 	// named pipes '.Listen' halts randomly on windows 2012, disabling named pipes on windows and using file channel instead
-	if runtime.GOOS == "windows" {
-		log.Info("Not using named pipe on windows")
+	// On few mac2.metal instances, socket creation is getting blocked. Hence, permanently falling back to File based IPC for Darwin.
+	if runtime.GOOS == "windows" || runtime.GOOS == "darwin" {
+		log.Infof("Not using named pipe on %v", runtime.GOOS)
 		return false
 	}
-
 	if appConfig.Agent.ForceFileIPC {
 		log.Info("Not using named pipe as force file IPC is set")
 		return false

common/channel/channel_darwin_test.go

@@ -0,0 +1,58 @@
+// Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"). You may not
+// use this file except in compliance with the License. A copy of the
+// License is located at
+//
+// http://aws.amazon.com/apache2.0/
+//
+// or in the "license" file accompanying this file. This file 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.
+//
+//go:build darwin
+// +build darwin
+
+// Package channel captures IPC implementation.
+package channel
+
+import (
+	"testing"
+
+	"github.com/aws/amazon-ssm-agent/agent/appconfig"
+	"github.com/aws/amazon-ssm-agent/agent/log"
+	logmocks "github.com/aws/amazon-ssm-agent/agent/mocks/log"
+	"github.com/aws/amazon-ssm-agent/common/channel/mocks"
+	"github.com/aws/amazon-ssm-agent/common/identity"
+	identityMocks "github.com/aws/amazon-ssm-agent/common/identity/mocks"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
+)
+
+type IChannelTestSuite struct {
+	suite.Suite
+	log       log.T
+	identity  identity.IAgentIdentity
+	appconfig appconfig.SsmagentConfig
+}
+
+// TestChannelSuite executes test suite
+func TestChannelSuite(t *testing.T) {
+	suite.Run(t, new(IChannelTestSuite))
+}
+
+// SetupTest initializes Setup
+func (suite *IChannelTestSuite) SetupTest() {
+	suite.log = logmocks.NewMockLog()
+	suite.identity = identityMocks.NewDefaultMockAgentIdentity()
+	suite.appconfig = appconfig.DefaultConfig()
+}
+
+func (suite *IChannelTestSuite) TestIPCSelection() {
+	newNamedPipeChannelRef = func(log log.T, identity identity.IAgentIdentity) IChannel {
+		return &mocks.IChannel{}
+	}
+	output := canUseNamedPipe(suite.log, suite.appconfig, suite.identity)
+	assert.False(suite.T(), output)
+}

common/channel/channel_unix_test.go

@@ -11,8 +11,8 @@
 // either express or implied. See the License for the specific language governing
 // permissions and limitations under the License.
 //
-//go:build darwin || freebsd || linux || netbsd || openbsd
-// +build darwin freebsd linux netbsd openbsd
+//go:build freebsd || linux || netbsd || openbsd
+// +build freebsd linux netbsd openbsd
 
 // Package channel captures IPC implementation.
 package channel

common/channel/filechannel.go

@@ -24,7 +24,13 @@ import (
 	"github.com/aws/amazon-ssm-agent/common/message"
 )
 
+var (
+	getSurveyInstance     = commProtocol.GetSurveyInstance
+	getRespondentInstance = commProtocol.GetRespondentInstance
+)
+
 // NewFileChannel creates an new instance of FileChannel which internally uses file watcher based ipc channel
+// This channel does not have multi-threading support. Currently, the invocations happen only in one go-routine
 func NewFileChannel(log log.T, identity identity.IAgentIdentity) IChannel {
 	return &fileChannel{
 		log:      log,
@@ -38,25 +44,35 @@ type fileChannel struct {
 	log                      log.T
 	identity                 identity.IAgentIdentity
 	isFileChannelInitialized bool
+	isDialSuccessful         bool
+	isListenSuccessful       bool
 	msgProtocol              utils.IFileChannelCommProtocol
 }
 
 // Initialize initializes survey properties
 func (fc *fileChannel) Initialize(socketType utils.SocketType) error {
 	fc.log.Info("using file channel for IPC")
 	if socketType == utils.Surveyor {
-		fc.msgProtocol = commProtocol.GetSurveyInstance(fc.log, fc.identity)
+		fc.msgProtocol = getSurveyInstance(fc.log, fc.identity)
 	} else if socketType == utils.Respondent {
-		fc.msgProtocol = commProtocol.GetRespondentInstance(fc.log, fc.identity)
+		fc.msgProtocol = getRespondentInstance(fc.log, fc.identity)
 	} else {
 		return fmt.Errorf("unsupported socket type")
 	}
 	fc.isFileChannelInitialized = true
+	fc.isDialSuccessful = false
+	fc.isListenSuccessful = false
 	return nil
 }
 
 // Send puts the message on the outbound send queue.
 func (fc *fileChannel) Send(message *message.Message) error {
+	if !fc.IsChannelInitialized() {
+		return ErrIPCChannelClosed
+	}
+	if !(fc.IsListenSuccessful() || fc.IsDialSuccessful()) {
+		return ErrDialListenUnSuccessful
+	}
 	return fc.msgProtocol.Send(message)
 }
 
@@ -69,6 +85,12 @@ func (fc *fileChannel) Close() error {
 // Recv gets the message from the IPC file channel created
 // message is returned whenever the worker creates a new file
 func (fc *fileChannel) Recv() ([]byte, error) {
+	if !fc.IsChannelInitialized() {
+		return nil, ErrIPCChannelClosed
+	}
+	if !(fc.IsListenSuccessful() || fc.IsDialSuccessful()) {
+		return nil, ErrDialListenUnSuccessful
+	}
 	return fc.msgProtocol.Recv()
 }
 
@@ -79,14 +101,37 @@ func (fc *fileChannel) SetOption(name string, value interface{}) (err error) {
 
 // Listen creates a new channel in the main worker side
 func (fc *fileChannel) Listen(addr string) error {
-	return fc.msgProtocol.Listen(addr)
+	err := fc.msgProtocol.Listen(addr)
+	if err != nil {
+		return err
+	}
+	fc.isListenSuccessful = true
+	return nil
 }
 
 // Dial creates a new channel in the worker side
 func (fc *fileChannel) Dial(addr string) error {
-	return fc.msgProtocol.Dial(addr)
+	err := fc.msgProtocol.Dial(addr)
+	if err != nil {
+		return err
+	}
+	fc.isDialSuccessful = true
+	return nil
 }
 
-func (fc *fileChannel) IsConnect() bool {
+// IsChannelInitialized returns true if channel initialization is successful.
+func (fc *fileChannel) IsChannelInitialized() bool {
 	return fc.isFileChannelInitialized
 }
+
+// IsDialSuccessful returns true if Dial() successfully connects to IPC channels.
+// In Dial(), we connect to the IPC with address being the parameter.
+func (fc *fileChannel) IsDialSuccessful() bool {
+	return fc.isDialSuccessful
+}
+
+// IsListenSuccessful returns true if Listen() successfully creates IPC channels.
+// In Listen(), we will create named pipes on Windows and sockets on Linux/Darwin for IPC.
+func (fc *fileChannel) IsListenSuccessful() bool {
+	return fc.isListenSuccessful
+}