Merge pull request #164 from utkarsh-pro/utkarsh-pro/fix/apply-config…

…-timeout

Fix #160

go.sum

@@ -272,6 +272,8 @@ github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
 github.com/go-ini/ini v1.25.4/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
+github.com/go-ini/ini v1.62.0 h1:7VJT/ZXjzqSrvtraFp4ONq80hTcRQth1c9ZnQ3uNQvU=
+github.com/go-ini/ini v1.62.0/go.mod h1:ByCAeIL28uOIIG0E3PJtZPDL8WnHpFKFOtgjp+3Ies8=
 github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
 github.com/go-kit/kit v0.10.0 h1:dXFJfIHVvUcpSgDOV+Ne6t7jXri8Tfv2uOLHUZ2XNuo=
@@ -1294,8 +1296,6 @@ gopkg.in/gorp.v1 v1.7.2 h1:j3DWlAyGVv8whO7AcIWznQ2Yj7yJkn34B8s63GViAAw=
 gopkg.in/gorp.v1 v1.7.2/go.mod h1:Wo3h+DBQZIxATwftsglhdD/62zRFPhGhTiu5jUJmCaw=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
-gopkg.in/ini.v1 v1.51.0 h1:AQvPpx3LzTDM0AjnIRlVFwFFGC+npRopjZxLJj6gdno=
-gopkg.in/ini.v1 v1.51.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
 gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
 gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
 gopkg.in/square/go-jose.v2 v2.2.2/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI=

istio/error.go

@@ -62,6 +62,10 @@ var (
 	// during istio-vet sync process
 	ErrIstioVetSyncCode = "istio_test_code"
 
+	// ErrIstioVetCode represents the errors which are generated
+	// during istio-vet process
+	ErrIstioVetCode = "istio_test_code"
+
 	// ErrOpInvalid represents the errors which are generated
 	// when an invalid operation is requested
 	ErrOpInvalid = errors.NewDefault(errors.ErrOpInvalid, "Invalid operation")
@@ -131,3 +135,8 @@ func ErrCreatingIstioClient(err error) error {
 func ErrIstioVetSync(err error) error {
 	return errors.NewDefault(ErrIstioVetSyncCode, fmt.Sprintf("Failed to sync %s", err.Error()))
 }
+
+// ErrIstioVet is the error for streaming event
+func ErrIstioVet(err error) error {
+	return errors.NewDefault(ErrIstioVetCode, err.Error())
+}

istio/istio.go

@@ -142,9 +142,9 @@ func (istio *Istio) ApplyOperation(ctx context.Context, opReq adapter.OperationR
 			for msg := range responseChan {
 				switch msg.EType {
 				case int32(meshes.EventType_ERROR):
-					istio.StreamErr(msg, fmt.Errorf(msg.Details))
+					istio.StreamErr(msg, ErrIstioVet(fmt.Errorf(msg.Details)))
 				case int32(meshes.EventType_WARN):
-					istio.StreamWarn(msg, fmt.Errorf(msg.Details))
+					istio.StreamWarn(msg, ErrIstioVet(fmt.Errorf(msg.Details)))
 				default:
 					istio.StreamInfo(msg)
 				}

istio/vet.go

@@ -2,7 +2,9 @@ package istio
 
 import (
 	"fmt"
+	"reflect"
 	"strings"
+	"time"
 
 	"github.com/aspenmesh/istio-vet/pkg/istioclient"
 	"github.com/aspenmesh/istio-vet/pkg/vetter"
@@ -19,6 +21,8 @@ import (
 	"k8s.io/client-go/informers"
 )
 
+const istioVetSyncTimeout = 10 // istio vet sync timeout in seconds
+
 type metaInformerFactory struct {
 	k8s   informers.SharedInformerFactory
 	istio istioinformer.SharedInformerFactory
@@ -33,6 +37,8 @@ func (m *metaInformerFactory) Istio() istioinformer.SharedInformerFactory {
 
 // RunVet runs istio-vet
 func (istio *Istio) RunVet(ch chan<- *adapter.Event) {
+	defer close(ch)
+
 	istioClient, err := istioclient.New(&istio.RestConfig)
 	if err != nil {
 		e := &adapter.Event{}
@@ -62,7 +68,18 @@ func (istio *Istio) RunVet(ch chan<- *adapter.Event) {
 	stopCh := make(chan struct{})
 
 	kubeInformerFactory.Start(stopCh)
-	oks := kubeInformerFactory.WaitForCacheSync(stopCh)
+	oks, timedout := completeBefore(istioVetSyncTimeout, func() map[reflect.Type]bool {
+		return kubeInformerFactory.WaitForCacheSync(stopCh)
+	})
+	if timedout {
+		e := &adapter.Event{}
+		e.EType = int32(meshes.EventType_ERROR)
+		e.Details = ErrIstioVetSync(fmt.Errorf("istio service mesh was either not found or is not deployed")).Error()
+		e.Summary = "Failed to sync: Request timed out"
+		ch <- e
+		close(stopCh)
+		return
+	}
 	for inf, ok := range oks {
 		if !ok {
 			e := &adapter.Event{}
@@ -75,7 +92,18 @@ func (istio *Istio) RunVet(ch chan<- *adapter.Event) {
 	}
 
 	istioInformerFactory.Start(stopCh)
-	oks = istioInformerFactory.WaitForCacheSync(stopCh)
+	oks, timedout = completeBefore(istioVetSyncTimeout, func() map[reflect.Type]bool {
+		return istioInformerFactory.WaitForCacheSync(stopCh)
+	})
+	if timedout {
+		e := &adapter.Event{}
+		e.EType = int32(meshes.EventType_ERROR)
+		e.Details = ErrIstioVetSync(fmt.Errorf("istio service mesh was either not found or is not deployed")).Error()
+		e.Summary = "Failed to sync: Request timed out"
+		ch <- e
+		close(stopCh)
+		return
+	}
 	for inf, ok := range oks {
 		if !ok {
 			e := &adapter.Event{}
@@ -128,8 +156,6 @@ func (istio *Istio) RunVet(ch chan<- *adapter.Event) {
 			ch <- e
 		}
 	}
-
-	close(ch)
 }
 
 // StreamWarn streams a warning message to the channel
@@ -138,3 +164,26 @@ func (istio *Istio) StreamWarn(e *adapter.Event, err error) {
 	e.EType = int32(meshes.EventType_WARN)
 	*istio.Channel <- e
 }
+
+// completeBefore executes the callback function but if the callback function
+// doesn't returns before the specified timeout then it returns nil and true
+// indicating that the request has timed out
+func completeBefore(timeout time.Duration, cb func() map[reflect.Type]bool) (map[reflect.Type]bool, bool) {
+	tch := make(chan bool, timeout)
+	resch := make(chan map[reflect.Type]bool)
+
+	go func() {
+		resch <- cb()
+	}()
+	go func() {
+		time.Sleep(timeout * time.Second)
+		tch <- true
+	}()
+
+	select {
+	case res := <-resch:
+		return res, false
+	case <-tch:
+		return nil, true
+	}
+}