update chart version to v1.19.0-rc2 (#2791)

Signed-off-by: Shubham Bansal <shubhambansaliimtgn@gmail.com>

charts/fission-all/Chart.yaml

@@ -1,7 +1,7 @@
 apiVersion: v2
 name: fission-all
-version: v1.19.0-rc1
-appVersion: v1.19.0-rc1
+version: v1.19.0-rc2
+appVersion: v1.19.0-rc2
 description: Fission is a fast serverless framework for Kubernetes.
 home: https://fission.io/
 icon: https://fission.io/images/fission-logo-white.svg

charts/fission-all/values.yaml

@@ -25,7 +25,7 @@ image: fission/fission-bundle
 ## It is also used by the chart to identify version of the few more images apart from fission-bundle.
 ## Keep it empty for using latest tag.
 ##
-imageTag: v1.19.0-rc1
+imageTag: v1.19.0-rc2
 
 ## pullPolicy represents the pull policy to use for images in the chart.
 ##
@@ -103,7 +103,7 @@ fetcher:
   ## image represents the image of the fetcher component.
   image: fission/fetcher
   ## imageTag represents the tag of the image of the fetcher component.
-  imageTag: v1.19.0-rc1
+  imageTag: v1.19.0-rc2
 
   ## Fetcher is only for to downloading or uploading archive.
   ## Normally, you don't need to change the value here, unless necessary.
@@ -685,7 +685,7 @@ preUpgradeChecks:
   image: fission/pre-upgrade-checks
   ## pre-install/pre-upgrade checks image version
   ##
-  imageTag: v1.19.0-rc1
+  imageTag: v1.19.0-rc2
 
 ## Fission post-install/post-upgrade reporting live in this image
 ##

pkg/builder/client/client.go

@@ -62,7 +62,7 @@ func (c *Client) Build(ctx context.Context, req *builder.PackageBuildRequest) (*
 
 	resp, err := ctxhttp.Post(ctx, c.httpClient.StandardClient(), c.url, "application/json", bytes.NewReader(body))
 	if err != nil {
-		return nil, ferror.MakeErrorFromHTTP(resp)
+		return nil, err
 	}
 	defer resp.Body.Close()
 

pkg/executor/api.go

@@ -31,8 +31,10 @@ import (
 	"go.uber.org/zap"
 
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
+	"github.com/fission/fission/pkg/crd"
 	ferror "github.com/fission/fission/pkg/error"
 	"github.com/fission/fission/pkg/executor/client"
+	"github.com/fission/fission/pkg/executor/fscache"
 	"github.com/fission/fission/pkg/utils/httpserver"
 	"github.com/fission/fission/pkg/utils/metrics"
 	otelUtils "github.com/fission/fission/pkg/utils/otel"
@@ -148,7 +150,24 @@ func (executor *Executor) getServiceForFunction(ctx context.Context, fn *fv1.Fun
 		respChan: respChan,
 	}
 	resp := <-respChan
+	cleanUp := func(funcSvc *fscache.FuncSvc) {
+		et, ok := executor.executorTypes[fn.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType]
+		if !ok {
+			executor.logger.Error("unknown executor type received in function service", zap.Any("executor", funcSvc.Executor))
+			return
+		}
+		if funcSvc != nil {
+			et.UnTapService(ctx, crd.CacheKey(funcSvc.Function), resp.funcSvc.Address)
+		} else {
+			et.MarkSpecializationFailure(ctx, crd.CacheKey(&fn.ObjectMeta))
+		}
+	}
+	if errors.Is(ctx.Err(), context.Canceled) {
+		cleanUp(resp.funcSvc)
+		return "", ferror.MakeError(499, "client leave early in the process of getServiceForFunction")
+	}
 	if resp.err != nil {
+		cleanUp(resp.funcSvc)
 		return "", resp.err
 	}
 	return resp.funcSvc.Address, resp.err
@@ -244,6 +263,19 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusOK)
 }
 
+// dumpDebugInfo => dump function service for pool cache
+func (executor *Executor) dumpDebugInfo(w http.ResponseWriter, r *http.Request) {
+	// currently we are considering dumping function only for pool manager
+	et := executor.executorTypes[fv1.ExecutorTypePoolmgr]
+	if err := et.DumpDebugInfo(r.Context()); err != nil {
+		code, msg := ferror.GetHTTPError(err)
+		http.Error(w, msg, code)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}
+
 // GetHandler returns an http.Handler.
 func (executor *Executor) GetHandler() http.Handler {
 	r := mux.NewRouter()
@@ -253,6 +285,7 @@ func (executor *Executor) GetHandler() http.Handler {
 	r.HandleFunc("/v2/tapServices", executor.tapServices).Methods("POST")
 	r.HandleFunc("/healthz", executor.healthHandler).Methods("GET")
 	r.HandleFunc("/v2/unTapService", executor.unTapService).Methods("POST")
+	r.HandleFunc("/v2/debugInfo", executor.dumpDebugInfo).Methods("GET")
 	return r
 }
 

pkg/executor/executortype/container/containermgr.go

@@ -177,6 +177,11 @@ func (caaf *Container) UnTapService(ctx context.Context, key string, svcHost str
 	// Not Implemented for CaaF.
 }
 
+// MarkSpecializationFailure has not been implemented for CaaF.
+func (caaf *Container) MarkSpecializationFailure(ctx context.Context, key string) {
+	// Not Implemented for CaaF.
+}
+
 // GetFuncSvc returns a function service; error otherwise.
 func (caaf *Container) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
 	return caaf.createFunction(ctx, fn)
@@ -457,7 +462,7 @@ func (caaf *Container) fnCreate(ctx context.Context, fn *fv1.Function) (*fscache
 	_, err = caaf.fsCache.Add(*fsvc)
 	if err != nil {
 		caaf.logger.Error("error adding function to cache", zap.Error(err), zap.Any("function", fsvc.Function))
-		metrics.FuncError.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
+		metrics.ColdStartsError.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
 		return fsvc, err
 	}
 
@@ -782,3 +787,7 @@ func getDeploymentObj(kubeobjs []apiv1.ObjectReference) *apiv1.ObjectReference {
 	}
 	return nil
 }
+
+func (caaf *Container) DumpDebugInfo(ctx context.Context) error {
+	return nil
+}

pkg/executor/executortype/executortype.go

@@ -38,6 +38,9 @@ type ExecutorType interface {
 	// GetFuncSvcFromCache retrieves function service from cache.
 	GetFuncSvcFromCache(context.Context, *fv1.Function) (*fscache.FuncSvc, error)
 
+	// DumpDebugInfo dump function service cache to temporary directory of executor pod.
+	DumpDebugInfo(context.Context) error
+
 	// DeleteFuncSvcFromCache deletes function service entry in cache.
 	DeleteFuncSvcFromCache(context.Context, *fscache.FuncSvc)
 
@@ -48,6 +51,9 @@ type ExecutorType interface {
 	// UnTapService updates the isActive to false
 	UnTapService(ctx context.Context, key string, svcHost string)
 
+	// ReduceSpecializationInProgress updates the svcWaiting count in funcSvcGroup
+	MarkSpecializationFailure(ctx context.Context, key string)
+
 	// IsValid returns true if a function service is valid. Different executor types
 	// use distinct ways to examine the function service.
 	IsValid(context.Context, *fscache.FuncSvc) bool

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -199,6 +199,11 @@ func (deploy *NewDeploy) UnTapService(ctx context.Context, key string, svcHost s
 	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
 }
 
+// MarkSpecializationFailure has not been implemented for NewDeployment.
+func (deploy *NewDeploy) MarkSpecializationFailure(ctx context.Context, key string) {
+	// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
+}
+
 // TapService makes a TouchByAddress request to the cache.
 func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error {
 	otelUtils.SpanTrackEvent(ctx, "TapService")
@@ -500,7 +505,7 @@ func (deploy *NewDeploy) fnCreate(ctx context.Context, fn *fv1.Function) (*fscac
 	_, err = deploy.fsCache.Add(*fsvc)
 	if err != nil {
 		deploy.logger.Error("error adding function to cache", zap.Error(err), zap.Any("function", fsvc.Function))
-		metrics.FuncError.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
+		metrics.ColdStartsError.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
 		return fsvc, err
 	}
 
@@ -884,3 +889,7 @@ func (deploy *NewDeploy) scaleDeployment(ctx context.Context, deplNS string, dep
 	}, metav1.UpdateOptions{})
 	return err
 }
+
+func (deploy *NewDeploy) DumpDebugInfo(ctx context.Context) error {
+	return nil
+}

pkg/executor/executortype/newdeploy/newdeploymgr_test.go

@@ -66,7 +66,7 @@ func TestRefreshFuncPods(t *testing.T) {
 
 	nsResolver := utils.NamespaceResolver{
 		FunctionNamespace: functionNamespace,
-		BuiderNamespace:   builderNamespace,
+		BuilderNamespace:  builderNamespace,
 		DefaultNamespace:  defaultNamespace,
 	}
 	ndm.nsResolver = &nsResolver

pkg/executor/executortype/poolmgr/gp.go

@@ -47,7 +47,6 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 	"github.com/fission/fission/pkg/crd"
 	"github.com/fission/fission/pkg/executor/fscache"
-	"github.com/fission/fission/pkg/executor/metrics"
 	fetcherClient "github.com/fission/fission/pkg/fetcher/client"
 	fetcherConfig "github.com/fission/fission/pkg/fetcher/config"
 	"github.com/fission/fission/pkg/generated/clientset/versioned"
@@ -227,6 +226,14 @@ func (gp *GenericPool) updateCPUUtilizationSvc(ctx context.Context) {
 // returns the key and pod API object.
 func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]string) (string, *apiv1.Pod, error) {
 	startTime := time.Now()
+	podTimeout := startTime.Add(gp.podReadyTimeout)
+	deadline, ok := ctx.Deadline()
+	if ok {
+		deadline = deadline.Add(-1 * time.Second)
+		if deadline.Before(podTimeout) {
+			podTimeout = deadline
+		}
+	}
 	expoDelay := 100 * time.Millisecond
 	logger := otelUtils.LoggerWithTraceID(ctx, gp.logger)
 	if !cache.WaitForCacheSync(ctx.Done(), gp.readyPodListerSynced) {
@@ -235,10 +242,14 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin
 	}
 	for {
 		// Retries took too long, error out.
-		if time.Since(startTime) > gp.podReadyTimeout {
-			logger.Error("timed out waiting for pod", zap.Any("labels", newLabels), zap.Duration("timeout", gp.podReadyTimeout))
+		if time.Now().After(podTimeout) {
+			logger.Error("timed out waiting for pod", zap.Any("labels", newLabels), zap.Duration("timeout", podTimeout.Sub(startTime)))
 			return "", nil, errors.New("timeout: waited too long to get a ready pod")
 		}
+		if ctx.Err() != nil {
+			logger.Error("context canceled while waiting for pod", zap.Any("labels", newLabels), zap.Duration("timeout", podTimeout.Sub(startTime)))
+			return "", nil, fmt.Errorf("context canceled while waiting for pod: %w", ctx.Err())
+		}
 
 		var chosenPod *apiv1.Pod
 		var key string
@@ -517,7 +528,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 		// Remove old versions function pods
 		for _, pod := range podList.Items {
 			// Delete pod no matter what status it is
-			gp.kubernetesClient.CoreV1().Pods(gp.fnNamespace).Delete(ctx, pod.ObjectMeta.Name, metav1.DeleteOptions{}) //nolint errcheck
+			gp.kubernetesClient.CoreV1().Pods(gp.fnNamespace).Delete(ctx, pod.ObjectMeta.Name, metav1.DeleteOptions{}) // nolint errcheck
 		}
 	}
 
@@ -613,7 +624,6 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 	gp.fsCache.PodToFsvc.Store(pod.GetObjectMeta().GetName(), fsvc)
 	gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address})
 	gp.fsCache.AddFunc(ctx, *fsvc, fn.GetRequestPerPod())
-	metrics.ColdStarts.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
 
 	logger.Info("added function service",
 		zap.String("pod", pod.ObjectMeta.Name),

pkg/executor/executortype/poolmgr/gpm.go

@@ -26,6 +26,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/fission/fission/pkg/executor/metrics"
 	"github.com/hashicorp/go-multierror"
 	"go.opentelemetry.io/otel/attribute"
 	"go.uber.org/zap"
@@ -186,7 +187,16 @@ func (gpm *GenericPoolManager) GetTypeName(ctx context.Context) fv1.ExecutorType
 	return fv1.ExecutorTypePoolmgr
 }
 
-func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
+func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (fnSvc *fscache.FuncSvc, fErr error) {
+	defer func() {
+		if fErr != nil {
+			metrics.ColdStartsError.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
+			return
+		}
+
+		metrics.ColdStarts.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc()
+	}()
+
 	otelUtils.SpanTrackEvent(ctx, "GetFuncSvc", otelUtils.GetAttributesForFunction(fn)...)
 	logger := otelUtils.LoggerWithTraceID(ctx, gpm.logger)
 
@@ -194,12 +204,14 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function)
 	logger.Debug("getting environment for function", zap.String("function", fn.ObjectMeta.Name))
 	env, err := gpm.getFunctionEnv(ctx, fn)
 	if err != nil {
-		return nil, err
+		fErr = err
+		return
 	}
 
 	pool, created, err := gpm.getPool(ctx, env)
 	if err != nil {
-		return nil, err
+		fErr = err
+		return
 	}
 
 	if created {
@@ -209,7 +221,8 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function)
 	// from GenericPool -> get one function container
 	// (this also adds to the cache)
 	logger.Debug("getting function service from pool", zap.String("function", fn.ObjectMeta.Name))
-	return pool.getFuncSvc(ctx, fn)
+	fnSvc, fErr = pool.getFuncSvc(ctx, fn)
+	return fnSvc, fErr
 }
 
 func (gpm *GenericPoolManager) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) {
@@ -239,6 +252,14 @@ func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) e
 	return nil
 }
 
+func (gpm *GenericPoolManager) MarkSpecializationFailure(ctx context.Context, key string) {
+	otelUtils.SpanTrackEvent(ctx, "MarkSpecializationFailure",
+		attribute.KeyValue{Key: "key", Value: attribute.StringValue(key)})
+	logger := otelUtils.LoggerWithTraceID(ctx, gpm.logger)
+	logger.Info("marking specialization failure", zap.Any("key", key))
+	gpm.fsCache.MarkSpecializationFailure(key)
+}
+
 // IsValid checks if pod is not deleted and that it has the address passed as the argument. Also checks that all the
 // containers in it are reporting a ready status for the healthCheck.
 func (gpm *GenericPoolManager) IsValid(ctx context.Context, fsvc *fscache.FuncSvc) bool {
@@ -741,3 +762,7 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(ctx context.Contex
 	}
 	wg.Wait()
 }
+
+func (gpm *GenericPoolManager) DumpDebugInfo(ctx context.Context) error {
+	return gpm.fsCache.DumpDebugInfo(ctx)
+}

pkg/executor/fscache/functionServiceCache.go

@@ -35,6 +35,7 @@ import (
 	"github.com/fission/fission/pkg/crd"
 	ferror "github.com/fission/fission/pkg/error"
 	"github.com/fission/fission/pkg/executor/metrics"
+	"github.com/fission/fission/pkg/executor/util"
 )
 
 type fscRequestType int
@@ -170,6 +171,27 @@ func (fsc *FunctionServiceCache) service() {
 	}
 }
 
+// DumpDebugInfo => dump function service cache data to temporary directory of executor pod.
+func (fsc *FunctionServiceCache) DumpDebugInfo(ctx context.Context) error {
+	fsc.logger.Info("dumping function service")
+
+	file, err := util.CreateDumpFile(fsc.logger)
+	if err != nil {
+		fsc.logger.Error("error while creating file/dir", zap.String("error", err.Error()))
+		return err
+	}
+	defer file.Close()
+
+	err = fsc.connFunctionCache.LogFnSvcGroup(ctx, file)
+	if err != nil {
+		fsc.logger.Error("error while logging function service group", zap.String("error", err.Error()))
+		return err
+	}
+
+	fsc.logger.Info("dumped function service")
+	return nil
+}
+
 // GetByFunction gets a function service from cache using function key.
 func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) {
 	key := crd.CacheKey(m)
@@ -244,6 +266,10 @@ func (fsc *FunctionServiceCache) MarkAvailable(key string, svcHost string) {
 	fsc.connFunctionCache.MarkAvailable(key, svcHost)
 }
 
+func (fsc *FunctionServiceCache) MarkSpecializationFailure(key string) {
+	fsc.connFunctionCache.MarkSpecializationFailure(key)
+}
+
 // Add adds a function service to cache if it does not exist already.
 func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
 	existing, err := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc)

pkg/executor/fscache/poolcache.go

@@ -17,8 +17,10 @@ limitations under the License.
 package fscache
 
 import (
+	"bufio"
 	"context"
 	"fmt"
+	"io"
 
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -36,6 +38,8 @@ const (
 	markAvailable
 	deleteValue
 	setCPUUtilization
+	markSpecializationFailure
+	logFuncSvc
 )
 
 type (
@@ -65,6 +69,7 @@ type (
 		ctx             context.Context
 		function        string
 		address         string
+		dumpWriter      io.Writer
 		value           *FuncSvc
 		requestsPerPod  int
 		cpuUsage        resource.Quantity
@@ -232,9 +237,59 @@ func (c *PoolCache) service() {
 					}
 				}
 			}
+		case markSpecializationFailure:
+			if c.cache[req.function].svcWaiting > c.cache[req.function].queue.Len() {
+				c.cache[req.function].svcWaiting--
+				if c.cache[req.function].svcWaiting == c.cache[req.function].queue.Len() {
+					expiredRequests := c.cache[req.function].queue.Expired()
+					c.cache[req.function].svcWaiting = c.cache[req.function].svcWaiting - expiredRequests
+				}
+			}
 		case deleteValue:
 			delete(c.cache[req.function].svcs, req.address)
 			req.responseChannel <- resp
+		case logFuncSvc:
+			datawriter := bufio.NewWriter(req.dumpWriter)
+
+			writefnSvcGrp := func(svcGrp *funcSvcGroup) error {
+				_, err := datawriter.WriteString(fmt.Sprintf("svc_waiting:%d\tqueue_len:%d", svcGrp.svcWaiting, svcGrp.queue.Len()))
+				if err != nil {
+					return err
+				}
+
+				if len(svcGrp.svcs) == 0 {
+					_, err := datawriter.WriteString("\n")
+					if err != nil {
+						return err
+					}
+				}
+
+				for addr, fnSvc := range svcGrp.svcs {
+					_, err := datawriter.WriteString(fmt.Sprintf("\tfunction_name:%s\tfn_svc_address:%s\tactive_req:%d\tcurrent_cpu_usage:%v\tcpu_limit:%v\n",
+						fnSvc.val.Function.Name, addr, fnSvc.activeRequests, fnSvc.currentCPUUsage, fnSvc.cpuLimit))
+					if err != nil {
+						return err
+					}
+				}
+				return nil
+			}
+
+			for _, fnSvcGrp := range c.cache {
+				err := writefnSvcGrp(fnSvcGrp)
+				if err != nil {
+					resp.error = err
+					break
+				}
+			}
+			err := datawriter.Flush()
+			if err != nil {
+				if resp.error == nil {
+					resp.error = err
+				} else {
+					resp.error = fmt.Errorf("%v, %v", resp.error, err)
+				}
+			}
+			req.responseChannel <- resp
 		default:
 			resp.error = ferror.MakeError(ferror.ErrorInvalidArgument,
 				fmt.Sprintf("invalid request type: %v", req.requestType))
@@ -328,3 +383,23 @@ func (c *PoolCache) DeleteValue(ctx context.Context, function, address string) e
 	resp := <-respChannel
 	return resp.error
 }
+
+// ReduceSpecializationInProgress reduces the svcWaiting count
+func (c *PoolCache) MarkSpecializationFailure(function string) {
+	c.requestChannel <- &request{
+		requestType:     markSpecializationFailure,
+		function:        function,
+		responseChannel: make(chan *response),
+	}
+}
+
+func (c *PoolCache) LogFnSvcGroup(ctx context.Context, file io.Writer) error {
+	respChannel := make(chan *response)
+	c.requestChannel <- &request{
+		requestType:     logFuncSvc,
+		dumpWriter:      file,
+		responseChannel: respChannel,
+	}
+	resp := <-respChannel
+	return resp.error
+}

pkg/executor/fscache/queue.go

@@ -38,6 +38,31 @@ func (q *Queue) Pop() *svcWait {
 	return svcWait
 }
 
+func (q *Queue) Expired() int {
+	q.mutex.Lock()
+	defer q.mutex.Unlock()
+
+	expired := 0
+	svcExpired := []*list.Element{}
+	for item := q.items.Front(); item != nil; item = item.Next() {
+		svcWait, ok := item.Value.(*svcWait)
+		if !ok {
+			continue
+		}
+		if svcWait.ctx.Err() != nil {
+			close(svcWait.svcChannel)
+			svcExpired = append(svcExpired, item)
+			expired = expired + 1
+		}
+	}
+
+	for _, item := range svcExpired {
+		q.items.Remove(item)
+	}
+
+	return expired
+}
+
 func (q *Queue) Len() int {
 	q.mutex.Lock()
 	defer q.mutex.Unlock()

pkg/executor/fscache/queue_test.go

@@ -1,6 +1,7 @@
 package fscache
 
 import (
+	"context"
 	"sync"
 	"testing"
 )
@@ -113,3 +114,78 @@ func TestQueueLen(t *testing.T) {
 		t.Errorf("Expected queue length to be 1, got %d", q.Len())
 	}
 }
+
+func TestExpiredWhenAllItemsExpired(t *testing.T) {
+	q := NewQueue()
+	if q.Expired() != 0 {
+		t.Errorf("Expected Expired to return 0, got %d", q.Expired())
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	item := &svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        ctx,
+	}
+	q.Push(item)
+	if q.Len() != 1 {
+		t.Errorf("Expected queue length to be 1, got %d", q.Len())
+	}
+	cancel()
+	if q.Expired() != 1 {
+		t.Errorf("Expected Expired to return 1, got %d", q.Expired())
+	}
+	if q.Len() != 0 {
+		t.Errorf("Expected queue length to be 0, got %d", q.Len())
+	}
+}
+
+func TestExpiredWhenFewItemsExpired(t *testing.T) {
+	q := NewQueue()
+	if q.Expired() != 0 {
+		t.Errorf("Expected Expired to return 0, got %d", q.Expired())
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+
+	q.Push(&svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        ctx,
+	})
+	q.Push(&svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        context.Background(),
+	})
+	if q.Len() != 2 {
+		t.Errorf("Expected queue length to be 1, got %d", q.Len())
+	}
+	cancel()
+	if q.Expired() != 1 {
+		t.Errorf("Expected Expired to return 1, got %d", q.Expired())
+	}
+	if q.Len() != 1 {
+		t.Errorf("Expected queue length to be 0, got %d", q.Len())
+	}
+}
+
+func TestExpiredWhenNoItemsExpired(t *testing.T) {
+	q := NewQueue()
+	if q.Expired() != 0 {
+		t.Errorf("Expected Expired to return 0, got %d", q.Expired())
+	}
+
+	q.Push(&svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        context.Background(),
+	})
+	q.Push(&svcWait{
+		svcChannel: make(chan *FuncSvc),
+		ctx:        context.Background(),
+	})
+	if q.Len() != 2 {
+		t.Errorf("Expected queue length to be 1, got %d", q.Len())
+	}
+	if q.Expired() != 0 {
+		t.Errorf("Expected Expired to return 1, got %d", q.Expired())
+	}
+	if q.Len() != 2 {
+		t.Errorf("Expected queue length to be 0, got %d", q.Len())
+	}
+}

pkg/executor/metrics/metrics.go

@@ -41,7 +41,7 @@ var (
 		},
 		functionLabels,
 	)
-	FuncError = prometheus.NewCounterVec(
+	ColdStartsError = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "fission_function_cold_start_errors_total",
 			Help: "Count of fission cold start errors",
@@ -54,5 +54,5 @@ func init() {
 	registry := metrics.Registry
 	registry.MustRegister(ColdStarts)
 	registry.MustRegister(FuncRunningSummary)
-	registry.MustRegister(FuncError)
+	registry.MustRegister(ColdStartsError)
 }

pkg/executor/util/util.go

@@ -35,6 +35,10 @@ import (
 	"github.com/fission/fission/pkg/utils"
 )
 
+const (
+	dumpFileName string = "fission-dump"
+)
+
 // ApplyImagePullSecret applies image pull secret to the give pod spec.
 // It's intentional not to check the existence of secret here.
 // First, Kubernetes will set Pod status to "ImagePullBackOff" once
@@ -152,3 +156,11 @@ func GetObjectReaperInterval(logger *zap.Logger, executorType fv1.ExecutorType,
 func getExecutorEnvVarName(executor fv1.ExecutorType) string {
 	return strings.ToUpper(string(executor)) + "_OBJECT_REAPER_INTERVAL"
 }
+
+// CreateDumpFile => create dump file inside temp directory
+func CreateDumpFile(logger *zap.Logger) (*os.File, error) {
+	dumpPath := os.TempDir()
+	logger.Info("creating dump file", zap.String("dump_path", dumpPath))
+
+	return os.Create(fmt.Sprintf("%s/%s-%d.txt", dumpPath, dumpFileName, time.Now().Unix()))
+}

pkg/fission-cli/cmd/environment/create.go

@@ -100,7 +100,7 @@ func (opts *CreateSubCommand) run(input cli.Input) (err error) {
 		}
 
 		specFile := fmt.Sprintf("env-%v.yaml", m.Name)
-		err = spec.SpecSave(*opts.env, specFile)
+		err = spec.SpecSave(*opts.env, specFile, false)
 		if err != nil {
 			return errors.Wrap(err, "error saving environment spec")
 		}

pkg/fission-cli/cmd/environment/update.go

@@ -29,6 +29,7 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 	"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
 	"github.com/fission/fission/pkg/fission-cli/cmd"
+	"github.com/fission/fission/pkg/fission-cli/cmd/spec"
 	"github.com/fission/fission/pkg/fission-cli/console"
 	flagkey "github.com/fission/fission/pkg/fission-cli/flag/key"
 	"github.com/fission/fission/pkg/fission-cli/util"
@@ -78,7 +79,20 @@ func (opts *UpdateSubCommand) complete(input cli.Input) (err error) {
 }
 
 func (opts *UpdateSubCommand) run(input cli.Input) error {
+	m := opts.env.ObjectMeta
+	if input.Bool(flagkey.SpecSave) {
+		err := opts.env.Validate()
+		if err != nil {
+			return fv1.AggregateValidationErrors("Environment", err)
+		}
 
+		specFile := fmt.Sprintf("env-%s.yaml", m.Name)
+		err = spec.SpecSave(*opts.env, specFile, true)
+		if err != nil {
+			return errors.Wrap(err, "error saving environment spec")
+		}
+		return nil
+	}
 	enew, err := opts.Client().FissionClientSet.CoreV1().Environments(opts.env.ObjectMeta.Namespace).Update(input.Context(), opts.env, metav1.UpdateOptions{})
 	if err != nil {
 		return errors.Wrap(err, "error updating environment")

pkg/fission-cli/cmd/function/create.go

@@ -363,7 +363,7 @@ func (opts *CreateSubCommand) run(input cli.Input) error {
 	}
 
 	if input.Bool(flagkey.SpecSave) {
-		err := spec.SpecSave(*opts.function, opts.specFile)
+		err := spec.SpecSave(*opts.function, opts.specFile, false)
 		if err != nil {
 			return errors.Wrap(err, "error saving function spec")
 		}

pkg/fission-cli/cmd/function/run_container.go

@@ -226,7 +226,7 @@ func (opts *RunContainerSubCommand) run(input cli.Input) error {
 	}
 
 	if input.Bool(flagkey.SpecSave) {
-		err := spec.SpecSave(*opts.function, opts.specFile)
+		err := spec.SpecSave(*opts.function, opts.specFile, false)
 		if err != nil {
 			return errors.Wrap(err, "error saving function spec")
 		}

pkg/fission-cli/cmd/function/update.go

@@ -27,6 +27,7 @@ import (
 	"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
 	"github.com/fission/fission/pkg/fission-cli/cmd"
 	_package "github.com/fission/fission/pkg/fission-cli/cmd/package"
+	"github.com/fission/fission/pkg/fission-cli/cmd/spec"
 	"github.com/fission/fission/pkg/fission-cli/console"
 	flagkey "github.com/fission/fission/pkg/fission-cli/flag/key"
 	"github.com/fission/fission/pkg/fission-cli/util"
@@ -35,6 +36,7 @@ import (
 type UpdateSubCommand struct {
 	cmd.CommandActioner
 	function *fv1.Function
+	specFile string
 }
 
 func Update(input cli.Input) error {
@@ -55,6 +57,9 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error {
 	if err != nil {
 		return errors.Wrap(err, "error in updating function ")
 	}
+	if input.Bool(flagkey.SpecSave) {
+		opts.specFile = fmt.Sprintf("function-%s.yaml", fnName)
+	}
 
 	function, err := opts.Client().FissionClientSet.CoreV1().Functions(fnNamespace).Get(input.Context(), input.String(flagkey.FnName), metav1.GetOptions{})
 	if err != nil {
@@ -193,7 +198,7 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error {
 		return errors.Errorf("Package is used by multiple functions, use --%v to force update", flagkey.PkgForce)
 	}
 
-	newPkgMeta, err := _package.UpdatePackage(input, opts.Client(), pkg)
+	newPkgMeta, err := _package.UpdatePackage(input, opts.Client(), opts.specFile, pkg)
 	if err != nil {
 		return errors.Wrap(err, fmt.Sprintf("error updating package '%v'", pkgName))
 	}
@@ -243,6 +248,17 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error {
 }
 
 func (opts *UpdateSubCommand) run(input cli.Input) error {
+	if input.Bool(flagkey.SpecSave) {
+		err := opts.function.Validate()
+		if err != nil {
+			return fv1.AggregateValidationErrors("Function", err)
+		}
+		err = spec.SpecSave(*opts.function, opts.specFile, false)
+		if err != nil {
+			return errors.Wrap(err, "error saving function spec")
+		}
+		return nil
+	}
 	_, err := opts.Client().FissionClientSet.CoreV1().Functions(opts.function.Namespace).Update(input.Context(), opts.function, metav1.UpdateOptions{})
 	if err != nil {
 		return errors.Wrap(err, "error updating function")

pkg/fission-cli/cmd/httptrigger/create.go

@@ -219,7 +219,7 @@ func (opts *CreateSubCommand) run(input cli.Input) error {
 
 	if input.Bool(flagkey.SpecSave) {
 		specFile := fmt.Sprintf("route-%v.yaml", opts.trigger.ObjectMeta.Name)
-		err := spec.SpecSave(*opts.trigger, specFile)
+		err := spec.SpecSave(*opts.trigger, specFile, false)
 		if err != nil {
 			return errors.Wrap(err, "error saving HTTP trigger spec")
 		}

pkg/fission-cli/cmd/httptrigger/update.go

@@ -26,6 +26,7 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 	"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
 	"github.com/fission/fission/pkg/fission-cli/cmd"
+	"github.com/fission/fission/pkg/fission-cli/cmd/spec"
 	"github.com/fission/fission/pkg/fission-cli/console"
 	flagkey "github.com/fission/fission/pkg/fission-cli/flag/key"
 	"github.com/fission/fission/pkg/fission-cli/util"
@@ -148,7 +149,18 @@ func (opts *UpdateSubCommand) complete(input cli.Input) (err error) {
 }
 
 func (opts *UpdateSubCommand) run(input cli.Input) error {
-
+	if input.Bool(flagkey.SpecSave) {
+		err := opts.trigger.Validate()
+		if err != nil {
+			return fv1.AggregateValidationErrors("HTTPTrigger", err)
+		}
+		specFile := fmt.Sprintf("route-%s.yaml", opts.trigger.ObjectMeta.Name)
+		err = spec.SpecSave(*opts.trigger, specFile, true)
+		if err != nil {
+			return errors.Wrap(err, "error saving HTTP trigger spec")
+		}
+		return nil
+	}
 	err := util.CheckHTTPTriggerDuplicates(input.Context(), opts.Client(), opts.trigger)
 	if err != nil {
 		return errors.Wrap(err, "Error while creating HTTP Trigger")

pkg/fission-cli/cmd/kubewatch/create.go

@@ -119,7 +119,7 @@ func (opts *CreateSubCommand) run(input cli.Input) error {
 
 	if input.Bool(flagkey.SpecSave) {
 		specFile := fmt.Sprintf("kubewatch-%v.yaml", opts.watcher.ObjectMeta.Name)
-		err := spec.SpecSave(*opts.watcher, specFile)
+		err := spec.SpecSave(*opts.watcher, specFile, false)
 		if err != nil {
 			return errors.Wrap(err, "error saving kubewatch spec")
 		}

pkg/fission-cli/cmd/mqtrigger/create.go

@@ -203,7 +203,7 @@ func (opts *CreateSubCommand) run(input cli.Input) error {
 
 	if input.Bool(flagkey.SpecSave) {
 		specFile := fmt.Sprintf("mqtrigger-%v.yaml", opts.trigger.ObjectMeta.Name)
-		err := spec.SpecSave(*opts.trigger, specFile)
+		err := spec.SpecSave(*opts.trigger, specFile, false)
 		if err != nil {
 			return errors.Wrap(err, "error saving message queue trigger spec")
 		}

pkg/fission-cli/cmd/mqtrigger/update.go

@@ -25,6 +25,7 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 	"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
 	"github.com/fission/fission/pkg/fission-cli/cmd"
+	"github.com/fission/fission/pkg/fission-cli/cmd/spec"
 	"github.com/fission/fission/pkg/fission-cli/console"
 	flagkey "github.com/fission/fission/pkg/fission-cli/flag/key"
 	"github.com/fission/fission/pkg/fission-cli/util"
@@ -148,6 +149,18 @@ func (opts *UpdateSubCommand) complete(input cli.Input) (err error) {
 }
 
 func (opts *UpdateSubCommand) run(input cli.Input) error {
+	if input.Bool(flagkey.SpecSave) {
+		err := opts.trigger.Validate()
+		if err != nil {
+			return fv1.AggregateValidationErrors("MessageQueueTrigger", err)
+		}
+		specFile := fmt.Sprintf("mqtrigger-%s.yaml", opts.trigger.ObjectMeta.Name)
+		err = spec.SpecSave(*opts.trigger, specFile, true)
+		if err != nil {
+			return errors.Wrap(err, "error saving message queue trigger spec")
+		}
+		return nil
+	}
 	_, err := opts.Client().FissionClientSet.CoreV1().MessageQueueTriggers(opts.trigger.ObjectMeta.Namespace).Update(input.Context(), opts.trigger, metav1.UpdateOptions{})
 	if err != nil {
 		return errors.Wrap(err, "error updating message queue trigger")

pkg/fission-cli/cmd/package/create.go

@@ -216,7 +216,7 @@ func CreatePackage(input cli.Input, client cmd.Client, pkgName string, pkgNamesp
 			return &pkg.ObjectMeta, nil
 		}
 
-		err = spec.SpecSave(*pkg, specFile)
+		err = spec.SpecSave(*pkg, specFile, false)
 		if err != nil {
 			return nil, errors.Wrap(err, "error saving package spec")
 		}

pkg/fission-cli/cmd/package/package.go

@@ -174,7 +174,7 @@ func CreateArchive(client cmd.Client, input cli.Input, includeFiles []string, no
 				aus.Name = oldAus.Name
 			} else {
 				// save the uploadspec
-				err := spec.SpecSave(*aus, specFile)
+				err := spec.SpecSave(*aus, specFile, false)
 				if err != nil {
 					return nil, errors.Wrap(err, "error saving archive spec")
 				}

pkg/fission-cli/cmd/package/update.go

@@ -28,7 +28,9 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 	"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
 	"github.com/fission/fission/pkg/fission-cli/cmd"
+	"github.com/fission/fission/pkg/fission-cli/cmd/spec"
 	flagkey "github.com/fission/fission/pkg/fission-cli/flag/key"
+	"github.com/fission/fission/pkg/fission-cli/util"
 )
 
 type UpdateSubCommand struct {
@@ -61,6 +63,7 @@ func (opts *UpdateSubCommand) complete(input cli.Input) (err error) {
 }
 
 func (opts *UpdateSubCommand) run(input cli.Input) error {
+	pkgName := input.String(flagkey.PkgName)
 	pkg, err := opts.Client().FissionClientSet.CoreV1().Packages(opts.pkgNamespace).Get(input.Context(), opts.pkgName, metav1.GetOptions{})
 	if err != nil {
 		return err
@@ -79,8 +82,8 @@ func (opts *UpdateSubCommand) run(input cli.Input) error {
 	if !forceUpdate && len(fnList) > 1 {
 		return errors.Errorf("package is used by multiple functions, use --%v to force update", flagkey.PkgForce)
 	}
-
-	newPkgMeta, err := UpdatePackage(input, opts.Client(), pkg)
+	specFile := fmt.Sprintf("package-%s.yaml", pkgName)
+	newPkgMeta, err := UpdatePackage(input, opts.Client(), specFile, pkg)
 	if err != nil {
 		return errors.Wrap(err, "error updating package")
 	}
@@ -95,7 +98,7 @@ func (opts *UpdateSubCommand) run(input cli.Input) error {
 	return nil
 }
 
-func UpdatePackage(input cli.Input, client cmd.Client, pkg *fv1.Package) (*metav1.ObjectMeta, error) {
+func UpdatePackage(input cli.Input, client cmd.Client, specFile string, pkg *fv1.Package) (*metav1.ObjectMeta, error) {
 	envName := input.String(flagkey.PkgEnvironment)
 	srcArchiveFiles := input.StringSlice(flagkey.PkgSrcArchive)
 	deployArchiveFiles := input.StringSlice(flagkey.PkgDeployArchive)
@@ -174,6 +177,27 @@ func UpdatePackage(input cli.Input, client cmd.Client, pkg *fv1.Package) (*metav
 		}
 	}
 
+	if input.Bool(flagkey.SpecSave) {
+		// if a package with the same spec exists, don't create a new spec file
+		fr, err := spec.ReadSpecs(util.GetSpecDir(input), util.GetSpecIgnore(input), false)
+		if err != nil {
+			return nil, errors.Wrap(err, "error reading specs")
+		}
+
+		obj := fr.SpecExists(pkg, true, true)
+		if obj != nil {
+			pkg := obj.(*fv1.Package)
+			fmt.Printf("Re-using previously created package %s\n", pkg.ObjectMeta.Name)
+			return &pkg.ObjectMeta, nil
+		}
+
+		err = spec.SpecSave(*pkg, specFile, true)
+		if err != nil {
+			return nil, errors.Wrap(err, "error saving package spec")
+		}
+		return &pkg.ObjectMeta, nil
+	}
+
 	newPkgMeta, err := client.FissionClientSet.CoreV1().Packages(pkg.ObjectMeta.Namespace).Update(input.Context(), pkg, metav1.UpdateOptions{})
 	if err != nil {
 		return nil, errors.Wrap(err, "update package")

pkg/fission-cli/cmd/spec/spec.go

@@ -127,7 +127,7 @@ func MapKey(m *metav1.ObjectMeta) string {
 }
 
 // save saves object encoded value to spec file under given spec directory
-func save(data []byte, specDir string, specFile string) error {
+func save(data []byte, specDir string, specFile string, truncate bool) error {
 	// verify
 	if _, err := os.Stat(filepath.Join(specDir, "fission-deployment-config.yaml")); os.IsNotExist(err) {
 		return errors.Wrap(err, "Couldn't find specs, run `fission spec init` first")
@@ -137,6 +137,9 @@ func save(data []byte, specDir string, specFile string) error {
 	// check if the file is new
 	newFile := false
 	if _, err := os.Stat(filename); os.IsNotExist(err) {
+		if truncate {
+			return errors.Errorf("spec file does not exists")
+		}
 		newFile = true
 	}
 
@@ -147,11 +150,19 @@ func save(data []byte, specDir string, specFile string) error {
 	}
 	defer f.Close()
 
-	// if we're appending, add a yaml document separator
-	if !newFile {
-		_, err = f.Write([]byte("\n---\n"))
+	if truncate {
+		err = f.Truncate(0)
 		if err != nil {
-			return errors.Wrap(err, "couldn't write to spec file")
+			return errors.Wrap(err, "couldn't truncate the spec file")
+		}
+
+	} else {
+		// if we're appending, add a yaml document separator
+		if !newFile {
+			_, err = f.Write([]byte("\n---\n"))
+			if err != nil {
+				return errors.Wrap(err, "couldn't write to spec file")
+			}
 		}
 	}
 
@@ -164,7 +175,7 @@ func save(data []byte, specDir string, specFile string) error {
 }
 
 // called from `fission * create --spec`
-func SpecSave(resource interface{}, specFile string) error {
+func SpecSave(resource interface{}, specFile string, update bool) error {
 	var specDir = "specs"
 
 	meta, kind, data, err := crdToYaml(resource)
@@ -186,7 +197,11 @@ func SpecSave(resource interface{}, specFile string) error {
 		return errors.Errorf("same name resource (%v) already exists in namespace (%v)", meta.Name, meta.Namespace)
 	}
 
-	err = save(data, specDir, specFile)
+	truncate := false
+	if update {
+		truncate = true
+	}
+	err = save(data, specDir, specFile, truncate)
 	if err != nil {
 		return err
 	}

pkg/fission-cli/cmd/timetrigger/create.go

@@ -135,7 +135,7 @@ func (opts *CreateSubCommand) run(input cli.Input) error {
 
 	if input.Bool(flagkey.SpecSave) {
 		specFile := fmt.Sprintf("timetrigger-%v.yaml", opts.trigger.ObjectMeta.Name)
-		err := spec.SpecSave(*opts.trigger, specFile)
+		err := spec.SpecSave(*opts.trigger, specFile, false)
 		if err != nil {
 			return errors.Wrap(err, "error saving time trigger spec")
 		}

pkg/fission-cli/cmd/timetrigger/update.go

@@ -25,6 +25,7 @@ import (
 	fv1 "github.com/fission/fission/pkg/apis/core/v1"
 	"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
 	"github.com/fission/fission/pkg/fission-cli/cmd"
+	"github.com/fission/fission/pkg/fission-cli/cmd/spec"
 	"github.com/fission/fission/pkg/fission-cli/console"
 	flagkey "github.com/fission/fission/pkg/fission-cli/flag/key"
 	"github.com/fission/fission/pkg/fission-cli/util"
@@ -86,6 +87,18 @@ func (opts *UpdateSubCommand) complete(input cli.Input) error {
 }
 
 func (opts *UpdateSubCommand) run(input cli.Input) error {
+	if input.Bool(flagkey.SpecSave) {
+		err := opts.trigger.Validate()
+		if err != nil {
+			return fv1.AggregateValidationErrors("TimeTrigger", err)
+		}
+		specFile := fmt.Sprintf("timetrigger-%s.yaml", opts.trigger.ObjectMeta.Name)
+		err = spec.SpecSave(*opts.trigger, specFile, true)
+		if err != nil {
+			return errors.Wrap(err, "error saving time trigger spec")
+		}
+		return nil
+	}
 	_, err := opts.Client().FissionClientSet.CoreV1().TimeTriggers(opts.trigger.ObjectMeta.Namespace).Update(input.Context(), opts.trigger, metav1.UpdateOptions{})
 	if err != nil {
 		return errors.Wrap(err, "error updating Time trigger")

pkg/utils/metrics/http_metrics.go

@@ -67,6 +67,12 @@ func (rw *ResponseWriterWrapper) WriteHeader(statuscode int) {
 	rw.ResponseWriter.WriteHeader(statuscode)
 }
 
+func (rw *ResponseWriterWrapper) Flush() {
+	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
 func HTTPMetricMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if util.IsWebsocketRequest(r) {

pkg/utils/metrics/http_metrics_test.go

@@ -0,0 +1,98 @@
+package metrics
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var dataRow = []byte("I'm the data Row\n")
+
+func chunkedHandler(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithCancel(r.Context())
+	ticker := time.NewTicker(time.Second) // We may set it to 10 secs
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ticker.C:
+				_, _ = w.Write(dataRow)
+				if f, ok := w.(http.Flusher); ok {
+					f.Flush()
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	// Emulate some work
+	time.Sleep(5 * time.Second)
+
+	// Telling the loop that keeps the connection alive to end
+	cancel()
+
+	// Waiting until the loop ends
+	wg.Wait()
+}
+
+func TestChunked(t *testing.T) {
+	mr := mux.NewRouter()
+	mr.Use(HTTPMetricMiddleware)
+	mr.Handle("/", http.HandlerFunc(chunkedHandler))
+	s := httptest.NewServer(mr)
+	defer s.Close()
+
+	resp, err := http.Get(s.URL)
+	require.NoError(t, err)
+	assert.Contains(t, resp.TransferEncoding, "chunked")
+	defer resp.Body.Close()
+
+	r := bufio.NewReader(resp.Body)
+	for {
+		line, err := readChunkedResponseLine(r)
+		if err != nil {
+			if err == io.EOF {
+				return
+			}
+			log.Fatal(err.Error())
+		}
+		if len(line) == 0 {
+			log.Println("Alive!")
+			continue
+		}
+
+		fmt.Println(string(line)) // we got the final response
+		assert.Equal(t, dataRow, append(line, '\n'))
+	}
+}
+
+func readChunkedResponseLine(r *bufio.Reader) ([]byte, error) {
+	line, isPrefix, err := r.ReadLine()
+	if err != nil {
+		return nil, err
+	}
+
+	if isPrefix {
+		rest, err := readChunkedResponseLine(r)
+		if err != nil {
+			return nil, err
+		}
+		line = append(line, rest...)
+	}
+
+	return line, nil
+}

pkg/utils/namespace.go

@@ -20,7 +20,7 @@ const (
 type (
 	NamespaceResolver struct {
 		FunctionNamespace string
-		BuiderNamespace   string
+		BuilderNamespace  string
 		DefaultNamespace  string
 		FissionResourceNS map[string]string
 		Logger            *zap.Logger
@@ -40,14 +40,14 @@ var nsResolver *NamespaceResolver
 func init() {
 	nsResolver = &NamespaceResolver{
 		FunctionNamespace: os.Getenv(ENV_FUNCTION_NAMESPACE),
-		BuiderNamespace:   os.Getenv(ENV_BUILDER_NAMESPACE),
+		BuilderNamespace:  os.Getenv(ENV_BUILDER_NAMESPACE),
 		DefaultNamespace:  os.Getenv(ENV_DEFAULT_NAMESPACE),
 		FissionResourceNS: GetNamespaces(),
 		Logger:            loggerfactory.GetLogger(),
 	}
 
 	nsResolver.Logger.Debug("namespaces", zap.String("function_namespace", nsResolver.FunctionNamespace),
-		zap.String("builder_namespace", nsResolver.BuiderNamespace),
+		zap.String("builder_namespace", nsResolver.BuilderNamespace),
 		zap.String("default_namespace", nsResolver.DefaultNamespace),
 		zap.Any("fission_resource_namespace", listNamespaces(nsResolver.FissionResourceNS)))
 }
@@ -96,8 +96,8 @@ func (nsr *NamespaceResolver) FissionNSWithOptions(option ...option) map[string]
 	if options.functionNS && nsr.FunctionNamespace != "" {
 		fissionResourceNS[nsr.FunctionNamespace] = nsr.FunctionNamespace
 	}
-	if options.builderNS && nsr.BuiderNamespace != "" {
-		fissionResourceNS[nsr.BuiderNamespace] = nsr.BuiderNamespace
+	if options.builderNS && nsr.BuilderNamespace != "" {
+		fissionResourceNS[nsr.BuilderNamespace] = nsr.BuilderNamespace
 	}
 	if options.defaultNs && nsr.DefaultNamespace != "" {
 		fissionResourceNS[nsr.DefaultNamespace] = nsr.DefaultNamespace
@@ -118,7 +118,7 @@ func GetNamespaces() map[string]string {
 	if len(envValue) > 0 {
 		lstNamespaces := strings.Split(envValue, ",")
 		for _, namespace := range lstNamespaces {
-			//check to handle string with additional comma at the end of string. eg- ns1,ns2,
+			// check to handle string with additional comma at the end of string. eg- ns1,ns2,
 			if namespace != "" {
 				namespaces[namespace] = namespace
 			}
@@ -132,14 +132,14 @@ func GetNamespaces() map[string]string {
 }
 
 func (nsr *NamespaceResolver) GetBuilderNS(namespace string) string {
-	if nsr.BuiderNamespace == "" {
+	if nsr.BuilderNamespace == "" {
 		return namespace
 	}
 
 	if namespace != metav1.NamespaceDefault {
 		return namespace
 	}
-	return nsr.BuiderNamespace
+	return nsr.BuilderNamespace
 }
 
 func (nsr *NamespaceResolver) GetFunctionNS(namespace string) string {
@@ -154,7 +154,7 @@ func (nsr *NamespaceResolver) GetFunctionNS(namespace string) string {
 }
 
 func (nsr *NamespaceResolver) ResolveNamespace(namespace string) string {
-	if nsr.FunctionNamespace == "" || nsr.BuiderNamespace == "" {
+	if nsr.FunctionNamespace == "" || nsr.BuilderNamespace == "" {
 		return nsr.DefaultNamespace
 	}
 	return namespace

pkg/utils/namespace_test.go

@@ -226,7 +226,7 @@ func TestNamespaceResolver(t *testing.T) {
 func getFissionNamespaces(builderNS, functionNS, defaultNS string) *NamespaceResolver {
 	return &NamespaceResolver{
 		FunctionNamespace: functionNS,
-		BuiderNamespace:   builderNS,
+		BuilderNamespace:  builderNS,
 		DefaultNamespace:  defaultNS,
 	}
 }

pkg/utils/profile/profile.go

@@ -25,7 +25,6 @@ package profile
 
 import (
 	"context"
-	"fmt"
 	"net/http"
 	_ "net/http/pprof"
 	"os"
@@ -35,26 +34,18 @@ import (
 	"github.com/fission/fission/pkg/utils/httpserver"
 )
 
-func getPprofAddr() string {
-	pprofHost := os.Getenv("PPROF_HOST")
-	if pprofHost == "" {
-		pprofHost = "localhost"
-	}
-	pprofPort := os.Getenv("PPROF_PORT")
-	if pprofPort == "" {
-		pprofPort = "6060"
-	}
-	return fmt.Sprintf("%s:%s", pprofHost, pprofPort)
-}
-
 func ProfileIfEnabled(ctx context.Context, logger *zap.Logger) {
 	enablePprof := os.Getenv("PPROF_ENABLED")
 	if enablePprof != "true" {
 		return
 	}
+	pprofPort := os.Getenv("PPROF_PORT")
+	if pprofPort == "" {
+		pprofPort = "6060"
+	}
 
 	pprofMux := http.DefaultServeMux
 	http.DefaultServeMux = http.NewServeMux()
 
-	go httpserver.StartServer(ctx, logger, "pprof", getPprofAddr(), pprofMux)
+	go httpserver.StartServer(ctx, logger, "pprof", pprofPort, pprofMux)
 }