Spell fixes across code (#2288)

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## [v1.15.0](https://github.com/fission/fission/tree/v1.15.0) (2021-11-23)
+
+[Full Changelog](https://github.com/fission/fission/compare/v1.15.0-rc2...v1.15.0)
+
+**Merged pull requests:**
+
+- Update alpine base image to 3.14.3 and security fixes [\#2266](https://github.com/fission/fission/pull/2266) ([sanketsudake](https://github.com/sanketsudake))
+- Update chart logo with svg image [\#2264](https://github.com/fission/fission/pull/2264) ([sanketsudake](https://github.com/sanketsudake))
+
 ## [v1.15.0-rc2](https://github.com/fission/fission/tree/v1.15.0-rc2) (2021-11-11)
 
 [Full Changelog](https://github.com/fission/fission/compare/v1.15.0-rc1...v1.15.0-rc2)

Makefile

@@ -82,6 +82,13 @@ generate-swagger-doc:
 generate-cli-docs:
 	go run tools/cmd-docs/main.go -o "../fission.io/content/en/docs/fission-cli"
 
+generate-crd-ref-docs:
+	# crd-ref-docs: https://github.com/elastic/crd-ref-docs
+	crd-ref-docs --source-path=pkg/apis/core/v1 --config=tools/crd-ref-docs/config.yaml --renderer markdown
+	cp tools/crd-ref-docs/header.md crd_docs.md
+	cat out.md >> crd_docs.md && rm out.md
+	cp crd_docs.md ../fission.io/content/en/docs/crd/_index.md
+
 all-generators: codegen generate-crds generate-swagger-doc
 
 skaffold-prebuild:

charts/fission-all/Chart.yaml

@@ -1,7 +1,7 @@
 apiVersion: v2
 name: fission-all
-version: v1.15.0
-appVersion: v1.15.0
+version: v1.15.1
+appVersion: v1.15.1
 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.15.0
+imageTag: v1.15.1
 
 ## pullPolicy represents the pull policy to use for images in the chart.
 ##
@@ -85,7 +85,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.15.0
+  imageTag: v1.15.1
 
   ## Fetcher is only for to downloading or uploading archive.
   ## Normally, you don't need to change the value here, unless necessary.
@@ -451,7 +451,7 @@ preUpgradeChecks:
   image: fission/pre-upgrade-checks
   ## pre-install/pre-upgrade checks image version
   ##
-  imageTag: v1.15.0
+  imageTag: v1.15.1
 
 ## Fission post-install/post-upgrade reporting live in this image
 ##

cmd/fetcher/app/server.go

@@ -42,7 +42,7 @@ var (
 func Run(ctx context.Context, logger *zap.Logger) {
 	flag.Usage = fetcherUsage
 	collectorEndpoint := flag.String("jaeger-collector-endpoint", "", "")
-	specializeOnStart := flag.Bool("specialize-on-startup", false, "Flag to activate specialize process at pod starup")
+	specializeOnStart := flag.Bool("specialize-on-startup", false, "Flag to activate specialize process at pod startup")
 	specializePayload := flag.String("specialize-request", "", "JSON payload for specialize request")
 	secretDir := flag.String("secret-dir", "", "Path to shared secrets directory")
 	configDir := flag.String("cfgmap-dir", "", "Path to shared configmap directory")

cmd/preupgradechecks/checks.go

@@ -96,21 +96,21 @@ func (client *PreUpgradeTaskClient) LatestSchemaApplied(ctx context.Context) err
 	client.logger.Info("Checking if user has applied the latest CRDs")
 	funcCRD := client.GetFunctionCRD(ctx)
 	if funcCRD == nil {
-		return errors.New("Could not get the Function CRD")
+		return fmt.Errorf("could not get the Function CRD")
 	}
 	// Any new field added in Function spec can be checked here provided the substring matches the description in CRD Validation of the field
 	if !strings.Contains(funcCRD.Spec.String(), "RequestsPerPod") || !strings.Contains(funcCRD.Spec.String(), "OnceOnly") || !strings.Contains(funcCRD.Spec.String(), "PodSpec") {
-		return errors.New("Apply the newer CRDs before upgrading")
+		return fmt.Errorf("could not find RequestPerPod/OnceOnly/PodSpec in Function CRD")
 	}
 
 	mqtCRD := client.GetMqtCRD(ctx)
 	if mqtCRD == nil {
-		return errors.New("Could not get the MQT CRD")
+		return fmt.Errorf("could not get the MQT CRD")
 	}
 
 	// Any new field added in MQT spec can be checked here provided the substring matches the description in CRD Validation of the field
 	if !strings.Contains(mqtCRD.Spec.String(), "PodSpec") {
-		return errors.New("Apply the newer CRDs before upgrading")
+		return fmt.Errorf("could not find PodSpec field in MQT CRD")
 	}
 
 	return nil

cmd/preupgradechecks/main.go

@@ -67,7 +67,7 @@ Options:
 
 	err = crdBackedClient.LatestSchemaApplied(ctx)
 	if err != nil {
-		logger.Fatal("New CRDs are not applied")
+		logger.Fatal("New CRDs are not applied", zap.Error(err))
 	}
 	crdBackedClient.VerifyFunctionSpecReferences(ctx)
 }

crds/v1/fission.io_environments.yaml

@@ -9399,7 +9399,7 @@ spec:
                 format: int64
                 type: integer
               version:
-                description: "Version is the Environment API version \n Version \"1\" allows user to run code snippet in a file and it's supported by most of environments except tensorflow-serving. \n Version \"2\" supports downloading and compiling user function if source archive is not empty. \n Version \"3\" is almost the same with v2, but you're able to control the size of pre-warm pool of the environment."
+                description: "Version is the Environment API version \n Version \"1\" allows user to run code snippet in a file, and it's supported by most of the environments except tensorflow-serving. \n Version \"2\" supports downloading and compiling user function if source archive is not empty. \n Version \"3\" is almost the same with v2, but you're able to control the size of pre-warm pool of the environment."
                 type: integer
             required:
             - runtime

crds/v1/fission.io_functions.yaml

@@ -41,7 +41,7 @@ spec:
                     description: ExecutionStrategy specifies low-level parameters for function execution, such as the number of instances.
                     properties:
                       ExecutorType:
-                        description: "ExecutorType is the executor type of a function used. Defaults to \"poolmgr\". \n Available value:  - poolmgr  - newdeploy  - container"
+                        description: "ExecutorType is the executor type of function used. Defaults to \"poolmgr\". \n Available value:  - poolmgr  - newdeploy  - container"
                         type: string
                       MaxScale:
                         description: This is only for newdeploy to set up maximum replicas of deployment.
@@ -57,7 +57,7 @@ spec:
                         type: integer
                     type: object
                   StrategyType:
-                    description: StrategyType is the strategy type of a function. Now it only supports 'execution'.
+                    description: StrategyType is the strategy type of function. Now it only supports 'execution'.
                     type: string
                 type: object
               concurrency:

crds/v1/fission.io_httptriggers.yaml

@@ -33,7 +33,7 @@ spec:
             description: HTTPTriggerSpec is for router to expose user functions at the given URL path.
             properties:
               createingress:
-                description: If CreateIngress is true, router will create a ingress definition.
+                description: If CreateIngress is true, router will create an ingress definition.
                 type: boolean
               functionref:
                 description: FunctionReference is a reference to the target function.
@@ -58,12 +58,12 @@ spec:
                 description: 'TODO: remove this field since we have IngressConfig already Deprecated: the original idea of this field is not for setting Ingress. Since we have IngressConfig now, remove Host after couple releases.'
                 type: string
               ingressconfig:
-                description: 'TODO: make IngressConfig a independent Fission resource IngressConfig for router to set up Ingress.'
+                description: 'TODO: make IngressConfig an independent Fission resource IngressConfig for router to set up Ingress.'
                 properties:
                   annotations:
                     additionalProperties:
                       type: string
-                    description: Annotations will be add to metadata when creating Ingress.
+                    description: Annotations will be added to metadata when creating Ingress.
                     nullable: true
                     type: object
                   host:

crds/v1/fission.io_packages.yaml

@@ -50,7 +50,7 @@ spec:
                         type: string
                     type: object
                   literal:
-                    description: Literal contents of the package. Can be used for encoding packages below TODO (256KB?) size.
+                    description: Literal contents of the package. Can be used for encoding packages below TODO (256 KB?) size.
                     format: byte
                     type: string
                   type:
@@ -84,7 +84,7 @@ spec:
                         type: string
                     type: object
                   literal:
-                    description: Literal contents of the package. Can be used for encoding packages below TODO (256KB?) size.
+                    description: Literal contents of the package. Can be used for encoding packages below TODO (256 KB?) size.
                     format: byte
                     type: string
                   type:

go.mod

@@ -8,7 +8,6 @@ require (
 	github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 // indirect
 	github.com/Shopify/sarama v1.30.0
 	github.com/blend/go-sdk v1.20211025.3 // indirect
-	github.com/bsm/sarama-cluster v2.1.15+incompatible
 	github.com/containerd/continuity v0.2.1 // indirect
 	github.com/dchest/uniuri v0.0.0-20200228104902-7aecb25e1fe5
 	github.com/docker/go-connections v0.4.0 // indirect
@@ -36,7 +35,7 @@ require (
 	github.com/nats-io/stan.go v0.10.0
 	github.com/nwaples/rardecode v1.1.2 // indirect
 	github.com/opencontainers/image-spec v1.0.2 // indirect
-	github.com/opencontainers/runc v1.0.2 // indirect
+	github.com/opencontainers/runc v1.0.3 // indirect
 	github.com/ory/dockertest v3.3.5+incompatible
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0

go.sum

@@ -187,8 +187,6 @@ github.com/blend/sentry-go v1.0.1/go.mod h1:hgyX3WXen2YBiA0NitlfsXsvS+9ly2YlEBmm
 github.com/bmizerany/pat v0.0.0-20170815010413-6226ea591a40/go.mod h1:8rLXio+WjiTceGBHIoTvn60HIbs7Hm7bcHjyrSqYB9c=
 github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
 github.com/bonitoo-io/go-sql-bigquery v0.3.4-1.4.0/go.mod h1:J4Y6YJm0qTWB9aFziB7cPeSyc6dOZFyJdteSeybVpXQ=
-github.com/bsm/sarama-cluster v2.1.15+incompatible h1:RkV6WiNRnqEEbp81druK8zYhmnIgdOjqSVi0+9Cnl2A=
-github.com/bsm/sarama-cluster v2.1.15+incompatible/go.mod h1:r7ao+4tTNXvWm+VRpRJchr2kQhqxgmAp2iEX5W96gMM=
 github.com/c-bata/go-prompt v0.2.2/go.mod h1:VzqtzE2ksDBcdln8G7mk2RX9QyGjH+OVqOCSiVIqS34=
 github.com/cactus/go-statsd-client/statsd v0.0.0-20191106001114-12b4e2b38748/go.mod h1:l/bIBLeOl9eX+wxJAzxS4TveKRtAqlyDpHjhkfO0MEI=
 github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=
@@ -971,8 +969,8 @@ github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.0.2 h1:9yCKha/T5XdGtO0q9Q9a6T5NUCsTn/DrBg0D7ufOcFM=
 github.com/opencontainers/image-spec v1.0.2/go.mod h1:BtxoFyWECRxE4U/7sNtV5W15zMzWCbyJoFRP3s7yZA0=
-github.com/opencontainers/runc v1.0.2 h1:opHZMaswlyxz1OuGpBE53Dwe4/xF7EZTY0A2L/FpCOg=
-github.com/opencontainers/runc v1.0.2/go.mod h1:aTaHFFwQXuA71CiyxOdFFIorAoemI04suvGRQFzWTD0=
+github.com/opencontainers/runc v1.0.3 h1:1hbqejyQWCJBvtKAfdO0b1FmaEf2z/bxnjqbARass5k=
+github.com/opencontainers/runc v1.0.3/go.mod h1:aTaHFFwQXuA71CiyxOdFFIorAoemI04suvGRQFzWTD0=
 github.com/opencontainers/runtime-spec v1.0.3-0.20210326190908-1c3f411f0417/go.mod h1:jwyrGlmzljRJv/Fgzds9SsS/C5hL+LL3ko9hs6T5lQ0=
 github.com/opencontainers/selinux v1.8.2/go.mod h1:MUIHuUEvKB1wtJjQdOyYRgOnLD2xAPP8dBsCoU0KuF8=
 github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492/go.mod h1:Ngi6UdF0k5OKD5t5wlmGhe/EDKPoUM3BXZSSfIuJbis=

hack/update-swagger-docs.sh

@@ -18,7 +18,7 @@ kube::swagger::gen_types_swagger_doc() {
 
   echo "Generating swagger type docs for ${group_version} at ${gv_dir}"
 
-  sed 's/YEAR/2017/' hack/boilerplate.txt > "$TMPFILE"
+  # sed 's/YEAR/2017/' hack/boilerplate.txt > "$TMPFILE"
   echo "package ${group_version##*/}" >> "$TMPFILE"
   cat >> "$TMPFILE" <<EOF
 // This file contains a collection of methods that can be used from go-restful to

pkg/apis/core/v1/doc.go

@@ -22,10 +22,6 @@ limitations under the License.
 // +groupName=fission.io
 // +groupGoName=core
 //
-// In order not to break the backward compatibility, keep coreV1 types stay
-// at "fission.io" group instead of moving them to "core.fission.io".
-// If the value of group is different from the one we register, the
-// CRD client will not be able to get anything from the API server.
 package v1
 
 const (

pkg/apis/core/v1/types.go

@@ -229,7 +229,7 @@ type (
 	// externally.
 	ArchiveType string
 
-	// Archive contains or references a collection of source or
+	// Archive contains or references a collection of sources or
 	// binary files.
 	Archive struct {
 		// Type defines how the package is specified: literal or URL.
@@ -240,7 +240,7 @@ type (
 		Type ArchiveType `json:"type,omitempty"`
 
 		// Literal contents of the package. Can be used for
-		// encoding packages below TODO (256KB?) size.
+		// encoding packages below TODO (256 KB?) size.
 		// +optional
 		Literal []byte `json:"literal,omitempty"`
 
@@ -254,7 +254,7 @@ type (
 		Checksum Checksum `json:"checksum,omitempty"`
 	}
 
-	// EnvironmentReference is a reference to a environment.
+	// EnvironmentReference is a reference to an environment.
 	EnvironmentReference struct {
 		Namespace string `json:"namespace"`
 		Name      string `json:"name"`
@@ -431,7 +431,7 @@ type (
 		// +optional
 		ExecutionStrategy ExecutionStrategy `json:"ExecutionStrategy"`
 
-		// StrategyType is the strategy type of a function.
+		// StrategyType is the strategy type of function.
 		// Now it only supports 'execution'.
 		// +optional
 		StrategyType StrategyType `json:"StrategyType"`
@@ -450,7 +450,7 @@ type (
 	// and resources allocated to the function pod.
 	ExecutionStrategy struct {
 
-		// ExecutorType is the executor type of a function used. Defaults to "poolmgr".
+		// ExecutorType is the executor type of function used. Defaults to "poolmgr".
 		//
 		// Available value:
 		//  - poolmgr
@@ -578,8 +578,8 @@ type (
 	EnvironmentSpec struct {
 		// Version is the Environment API version
 		//
-		// Version "1" allows user to run code snippet in a file and
-		// it's supported by most of environments except tensorflow-serving.
+		// Version "1" allows user to run code snippet in a file, and
+		// it's supported by most of the environments except tensorflow-serving.
 		//
 		// Version "2" supports downloading and compiling user function if source archive is not empty.
 		//
@@ -681,11 +681,11 @@ type (
 		// FunctionReference is a reference to the target function.
 		FunctionReference FunctionReference `json:"functionref"`
 
-		// If CreateIngress is true, router will create a ingress definition.
+		// If CreateIngress is true, router will create an ingress definition.
 		// +optional
 		CreateIngress bool `json:"createingress"`
 
-		// TODO: make IngressConfig a independent Fission resource
+		// TODO: make IngressConfig an independent Fission resource
 		// IngressConfig for router to set up Ingress.
 		// +optional
 		IngressConfig IngressConfig `json:"ingressconfig"`
@@ -693,7 +693,7 @@ type (
 
 	// IngressConfig is for router to set up Ingress.
 	IngressConfig struct {
-		// Annotations will be add to metadata when creating Ingress.
+		// Annotations will be added to metadata when creating Ingress.
 		// +optional
 		// +nullable
 		Annotations map[string]string `json:"annotations"`

pkg/apis/core/v1/validation.go

@@ -457,7 +457,7 @@ func (config IngressConfig) Validate() error {
 
 	// In Ingress, to accept requests from all host, the host field will
 	// be an empty string instead of "*" shown in kubectl. The router replaces
-	// the asterisk with "" when creating/updateing the Ingress, so here we
+	// the asterisk with "" when creating/updating the Ingress, so here we
 	// skip the check if the Host is equal to "*".
 	if len(config.Host) > 0 && config.Host != "*" {
 		if strings.Contains(config.Host, "*") {

pkg/apis/core/v1/zz_generated.swagger_doc_generated.go

@@ -1,18 +1,3 @@
-/*
-Copyright The Fission Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License 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.
-*/
 package v1
 
 // This file contains a collection of methods that can be used from go-restful to
@@ -26,9 +11,9 @@ package v1
 // Those methods can be generated by using hack/update-swagger-docs.sh
 // AUTO-GENERATED FUNCTIONS START HERE
 var map_Archive = map[string]string{
-	"":         "Archive contains or references a collection of source or binary files.",
+	"":         "Archive contains or references a collection of sources or binary files.",
 	"type":     "Type defines how the package is specified: literal or URL. Available value:\n - literal\n - url",
-	"literal":  "Literal contents of the package. Can be used for encoding packages below TODO (256KB?) size.",
+	"literal":  "Literal contents of the package. Can be used for encoding packages below TODO (256 KB?) size.",
 	"url":      "URL references a package.",
 	"checksum": "Checksum ensures the integrity of packages referenced by URL. Ignored for literals.",
 }
@@ -120,7 +105,7 @@ func (EnvironmentList) SwaggerDoc() map[string]string {
 }
 
 var map_EnvironmentReference = map[string]string{
-	"": "EnvironmentReference is a reference to a environment.",
+	"": "EnvironmentReference is a reference to an environment.",
 }
 
 func (EnvironmentReference) SwaggerDoc() map[string]string {
@@ -129,7 +114,7 @@ func (EnvironmentReference) SwaggerDoc() map[string]string {
 
 var map_EnvironmentSpec = map[string]string{
 	"":                             "EnvironmentSpec contains with builder, runtime and some other related environment settings.",
-	"version":                      "Version is the Environment API version\n\nVersion \"1\" allows user to run code snippet in a file and it's supported by most of environments except tensorflow-serving.\n\nVersion \"2\" supports downloading and compiling user function if source archive is not empty.\n\nVersion \"3\" is almost the same with v2, but you're able to control the size of pre-warm pool of the environment.",
+	"version":                      "Version is the Environment API version\n\nVersion \"1\" allows user to run code snippet in a file, and it's supported by most of the environments except tensorflow-serving.\n\nVersion \"2\" supports downloading and compiling user function if source archive is not empty.\n\nVersion \"3\" is almost the same with v2, but you're able to control the size of pre-warm pool of the environment.",
 	"runtime":                      "Runtime is configuration for running function, like container image etc.",
 	"builder":                      "(Optional) Builder is configuration for builder manager to launch environment builder to build source code into deployable binary.",
 	"allowedFunctionsPerContainer": "(Optional) defaults to 'single'. Fission workflow uses 'infinite' to load multiple functions in one function pod. Available value: - single - infinite",
@@ -147,7 +132,7 @@ func (EnvironmentSpec) SwaggerDoc() map[string]string {
 
 var map_ExecutionStrategy = map[string]string{
 	"":                      "ExecutionStrategy specifies low-level parameters for function execution, such as the number of instances.\n\nMinScale affects the cold start behavior for a function. If MinScale is 0 then the deployment is created on first invocation of function and is good for requests of asynchronous nature. If MinScale is greater than 0 then MinScale number of pods are created at the time of creation of function. This ensures faster response during first invocation at the cost of consuming resources.\n\nMaxScale is the maximum number of pods that function will scale to based on TargetCPUPercent and resources allocated to the function pod.",
-	"ExecutorType":          "ExecutorType is the executor type of a function used. Defaults to \"poolmgr\".\n\nAvailable value:\n - poolmgr\n - newdeploy\n - container",
+	"ExecutorType":          "ExecutorType is the executor type of function used. Defaults to \"poolmgr\".\n\nAvailable value:\n - poolmgr\n - newdeploy\n - container",
 	"MinScale":              "This is only for newdeploy to set up minimum replicas of deployment.",
 	"MaxScale":              "This is only for newdeploy to set up maximum replicas of deployment.",
 	"TargetCPUPercent":      "This is only for newdeploy to set up target CPU utilization of HPA.",
@@ -240,7 +225,7 @@ var map_HTTPTriggerSpec = map[string]string{
 	"method":        "Use Methods instead of Method. This field is going to be deprecated in a future release HTTP method to access a function.",
 	"methods":       "HTTP methods to access a function",
 	"functionref":   "FunctionReference is a reference to the target function.",
-	"createingress": "If CreateIngress is true, router will create a ingress definition.",
+	"createingress": "If CreateIngress is true, router will create an ingress definition.",
 	"ingressconfig": "IngressConfig for router to set up Ingress.",
 }
 
@@ -250,7 +235,7 @@ func (HTTPTriggerSpec) SwaggerDoc() map[string]string {
 
 var map_IngressConfig = map[string]string{
 	"":            "IngressConfig is for router to set up Ingress.",
-	"annotations": "Annotations will be add to metadata when creating Ingress.",
+	"annotations": "Annotations will be added to metadata when creating Ingress.",
 	"path":        "Path is for path matching. The format of path depends on what ingress controller you used.",
 	"host":        "Host is for ingress controller to apply rules. If host is empty or \"*\", the rule applies to all inbound HTTP traffic.",
 	"tls":         "TLS is for user to specify a Secret that contains TLS key and certificate. The domain name in the key and crt must match the value of Host field.",
@@ -263,7 +248,7 @@ func (IngressConfig) SwaggerDoc() map[string]string {
 var map_InvokeStrategy = map[string]string{
 	"":                  "InvokeStrategy is a set of controls over how the function executes. It affects the performance and resource usage of the function.\n\nAn InvokeStrategy is of one of two types: ExecutionStrategy, which controls low-level parameters such as which ExecutorType to use, when to autoscale, minimum and maximum number of running instances, etc. A higher-level AbstractInvokeStrategy will also be supported; this strategy would specify the target request rate of the function, the target latency statistics, and the target cost (in terms of compute resources).",
 	"ExecutionStrategy": "ExecutionStrategy specifies low-level parameters for function execution, such as the number of instances.",
-	"StrategyType":      "StrategyType is the strategy type of a function. Now it only supports 'execution'.",
+	"StrategyType":      "StrategyType is the strategy type of function. Now it only supports 'execution'.",
 }
 
 func (InvokeStrategy) SwaggerDoc() map[string]string {

pkg/buildermgr/pkgwatcher.go

@@ -165,7 +165,7 @@ func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) {
 			}
 
 			// Add the package getter rolebinding to builder sa
-			// we continue here if role binding was not setup successfully. this is because without this, the fetcher wont be able to fetch the source pkg into the container and
+			// we continue here if role binding was not setup successfully. this is because without this, the fetcher won't be able to fetch the source pkg into the container and
 			// the build will fail eventually
 			err := utils.SetupRoleBinding(ctx, pkgw.logger, pkgw.k8sClient, fv1.PackageGetterRB, pkg.ObjectMeta.Namespace, fv1.PackageGetterCR, fv1.ClusterRole, fv1.FissionBuilderSA, builderNs)
 			if err != nil {

pkg/canaryconfigmgr/canaryConfigMgr.go

@@ -77,7 +77,7 @@ func MakeCanaryConfigMgr(logger *zap.Logger, fissionClient *crd.FissionClient, k
 
 	_, err := url.Parse(prometheusSvc)
 	if err != nil {
-		return nil, errors.Errorf("prometheus service url not found/invalid, cant create canary config manager: %v", prometheusSvc)
+		return nil, errors.Errorf("prometheus service url not found/invalid, can't create canary config manager: %v", prometheusSvc)
 	}
 
 	promClient, err := MakePrometheusClient(logger, prometheusSvc)
@@ -140,7 +140,7 @@ func (canaryCfgMgr *canaryConfigMgr) addCanaryConfig(canaryConfig *fv1.CanaryCon
 	// for each canary config, create a ticker with increment interval
 	interval, err := time.ParseDuration(canaryConfig.Spec.WeightIncrementDuration)
 	if err != nil {
-		canaryCfgMgr.logger.Error("error parsing duration - cant proceed with this canaryConfig",
+		canaryCfgMgr.logger.Error("error parsing duration - can't proceed with this canaryConfig",
 			zap.Error(err),
 			zap.String("duration", canaryConfig.Spec.WeightIncrementDuration),
 			zap.String("name", canaryConfig.ObjectMeta.Name),
@@ -351,7 +351,7 @@ func (canaryCfgMgr *canaryConfigMgr) RollForwardOrBack(canaryConfig *fv1.CanaryC
 		err = canaryCfgMgr.updateCanaryConfigStatusWithRetries(canaryConfig.ObjectMeta.Name, canaryConfig.ObjectMeta.Namespace,
 			fv1.CanaryConfigStatusSucceeded)
 		if err != nil {
-			// cant do much after max retries other than logging it.
+			// can't do much after max retries other than logging it.
 			canaryCfgMgr.logger.Error("error updating canary config after max retries",
 				zap.Error(err),
 				zap.String("name", canaryConfig.ObjectMeta.Name),

pkg/controller/watchApi.go

@@ -134,7 +134,7 @@ func (a *API) WatchApiCreate(w http.ResponseWriter, r *http.Request) {
 	}
 
 	// TODO check for duplicate watches
-	// TODO check for duplicate watches -> we probably wont need it?
+	// TODO check for duplicate watches -> we probably won't need it?
 	// check if namespace exists, if not create it.
 	err = a.createNsIfNotExists(r.Context(), watch.ObjectMeta.Namespace)
 	if err != nil {

pkg/executor/executor.go

@@ -299,7 +299,7 @@ func StartExecutor(ctx context.Context, logger *zap.Logger, functionNamespace st
 		funcInformer, pkgInformer, envInformer,
 		gpmPodInformer, gpmRsInformer)
 	if err != nil {
-		return errors.Wrap(err, "pool manager creation faied")
+		return errors.Wrap(err, "pool manager creation failed")
 	}
 
 	ndmInformerFactory, err := utils.GetInformerFactoryByExecutor(kubernetesClient, fv1.ExecutorTypeNewdeploy, time.Minute*30)
@@ -315,7 +315,7 @@ func StartExecutor(ctx context.Context, logger *zap.Logger, functionNamespace st
 		funcInformer, envInformer,
 		ndmDeplInformer, ndmSvcInformer)
 	if err != nil {
-		return errors.Wrap(err, "new deploy manager creation faied")
+		return errors.Wrap(err, "new deploy manager creation failed")
 	}
 
 	cnmInformerFactory, err := utils.GetInformerFactoryByExecutor(kubernetesClient, fv1.ExecutorTypeContainer, time.Minute*30)
@@ -330,7 +330,7 @@ func StartExecutor(ctx context.Context, logger *zap.Logger, functionNamespace st
 		functionNamespace, executorInstanceID, funcInformer,
 		cnmDeplInformer, cnmSvcInformer)
 	if err != nil {
-		return errors.Wrap(err, "container manager creation faied")
+		return errors.Wrap(err, "container manager creation failed")
 	}
 
 	executorTypes := make(map[fv1.ExecutorType]executortype.ExecutorType)

pkg/executor/executortype/container/containermgr.go

@@ -664,7 +664,7 @@ func (caaf *Container) getObjName(fn *fv1.Function) string {
 			functionMetadata = functionMetadata + "-" + fn.ObjectMeta.Namespace
 		}
 	}
-	// contructed name should be 63 characters long, as it is a valid k8s name
+	// constructed name should be 63 characters long, as it is a valid k8s name
 	// functionMetadata should be 35 characters long, as we take 17 characters from functionUid
 	// with newdeploy 10 character prefix
 	return strings.ToLower(fmt.Sprintf("container-%s-%s", functionMetadata, uid))

pkg/executor/executortype/container/deployment.go

@@ -55,40 +55,12 @@ func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function
 	}
 
 	existingDepl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Get(ctx, deployName, metav1.GetOptions{})
-	if err == nil {
-		// Try to adopt orphan deployment created by the old executor.
-		if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID {
-			existingDepl.Annotations = deployment.Annotations
-			existingDepl.Labels = deployment.Labels
-			existingDepl.Spec.Template.Spec.Containers = deployment.Spec.Template.Spec.Containers
-			existingDepl.Spec.Template.Spec.ServiceAccountName = deployment.Spec.Template.Spec.ServiceAccountName
-			existingDepl.Spec.Template.Spec.TerminationGracePeriodSeconds = deployment.Spec.Template.Spec.TerminationGracePeriodSeconds
+	if err != nil && !k8s_err.IsNotFound(err) {
+		return nil, err
+	}
 
-			// Update with the latest deployment spec. Kubernetes will trigger
-			// rolling update if spec is different from the one in the cluster.
-			existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(ctx, existingDepl, metav1.UpdateOptions{})
-			if err != nil {
-				logger.Warn("error adopting cn", zap.Error(err),
-					zap.String("cn", deployName), zap.String("ns", deployNamespace))
-				return nil, err
-			}
-			// In this case, we just return without waiting for it for fast bootstraping.
-			return existingDepl, nil
-		}
-
-		if *existingDepl.Spec.Replicas < minScale {
-			err = cn.scaleDeployment(ctx, existingDepl.Namespace, existingDepl.Name, minScale)
-			if err != nil {
-				logger.Error("error scaling up function deployment", zap.Error(err), zap.String("function", fn.ObjectMeta.Name))
-				return nil, err
-			}
-		}
-		if existingDepl.Status.AvailableReplicas < minScale {
-			existingDepl, err = cn.waitForDeploy(ctx, existingDepl, minScale, specializationTimeout)
-		}
-
-		return existingDepl, err
-	} else if k8s_err.IsNotFound(err) {
+	// Create new deployment if one does not previously exist
+	if k8s_err.IsNotFound(err) {
 		depl, err := cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Create(ctx, deployment, metav1.CreateOptions{})
 		if err != nil {
 			if k8s_err.IsAlreadyExists(err) {
@@ -109,7 +81,39 @@ func (cn *Container) createOrGetDeployment(ctx context.Context, fn *fv1.Function
 		}
 		return depl, err
 	}
-	return nil, err
+
+	// Try to adopt orphan deployment created by the old executor.
+	if existingDepl.Annotations[fv1.EXECUTOR_INSTANCEID_LABEL] != cn.instanceID {
+		existingDepl.Annotations = deployment.Annotations
+		existingDepl.Labels = deployment.Labels
+		existingDepl.Spec.Template.Spec.Containers = deployment.Spec.Template.Spec.Containers
+		existingDepl.Spec.Template.Spec.ServiceAccountName = deployment.Spec.Template.Spec.ServiceAccountName
+		existingDepl.Spec.Template.Spec.TerminationGracePeriodSeconds = deployment.Spec.Template.Spec.TerminationGracePeriodSeconds
+
+		// Update with the latest deployment spec. Kubernetes will trigger
+		// rolling update if spec is different from the one in the cluster.
+		existingDepl, err = cn.kubernetesClient.AppsV1().Deployments(deployNamespace).Update(ctx, existingDepl, metav1.UpdateOptions{})
+		if err != nil {
+			logger.Warn("error adopting cn", zap.Error(err),
+				zap.String("cn", deployName), zap.String("ns", deployNamespace))
+			return nil, err
+		}
+		// In this case, we just return without waiting for it for fast bootstraping.
+		return existingDepl, nil
+	}
+
+	if *existingDepl.Spec.Replicas < minScale {
+		err = cn.scaleDeployment(ctx, existingDepl.Namespace, existingDepl.Name, minScale)
+		if err != nil {
+			logger.Error("error scaling up function deployment", zap.Error(err), zap.String("function", fn.ObjectMeta.Name))
+			return nil, err
+		}
+	}
+	if existingDepl.Status.AvailableReplicas < minScale {
+		existingDepl, err = cn.waitForDeploy(ctx, existingDepl, minScale, specializationTimeout)
+	}
+
+	return existingDepl, err
 }
 
 func (cn *Container) updateDeployment(ctx context.Context, deployment *appsv1.Deployment, ns string) error {
@@ -119,7 +123,7 @@ func (cn *Container) updateDeployment(ctx context.Context, deployment *appsv1.De
 
 func (cn *Container) deleteDeployment(ctx context.Context, ns string, name string) error {
 	// DeletePropagationBackground deletes the object immediately and dependent are deleted later
-	// DeletePropagationForeground not advisable; it marks for deleteion and API can still serve those objects
+	// DeletePropagationForeground not advisable; it marks for deletion and API can still serve those objects
 	deletePropagation := metav1.DeletePropagationBackground
 	return cn.kubernetesClient.AppsV1().Deployments(ns).Delete(ctx, name, metav1.DeleteOptions{
 		PropagationPolicy: &deletePropagation,

pkg/executor/executortype/newdeploy/newdeploy.go

@@ -174,7 +174,7 @@ func (deploy *NewDeploy) updateDeployment(ctx context.Context, deployment *appsv
 
 func (deploy *NewDeploy) deleteDeployment(ctx context.Context, ns string, name string) error {
 	// DeletePropagationBackground deletes the object immediately and dependent are deleted later
-	// DeletePropagationForeground not advisable; it marks for deleteion and API can still serve those objects
+	// DeletePropagationForeground not advisable; it marks for deletion and API can still serve those objects
 	deletePropagation := metav1.DeletePropagationBackground
 	return deploy.kubernetesClient.AppsV1().Deployments(ns).Delete(ctx, name, metav1.DeleteOptions{
 		PropagationPolicy: &deletePropagation,

pkg/executor/executortype/newdeploy/newdeploymgr.go

@@ -712,7 +712,7 @@ func (deploy *NewDeploy) getObjName(fn *fv1.Function) string {
 			functionMetadata = functionMetadata + "-" + fn.ObjectMeta.Namespace
 		}
 	}
-	// contructed name should be 63 characters long, as it is a valid k8s name
+	// constructed name should be 63 characters long, as it is a valid k8s name
 	// functionMetadata should be 35 characters long, as we take 17 characters from functionUid
 	// with newdeploy 10 character prefix
 	return strings.ToLower(fmt.Sprintf("newdeploy-%s-%s", functionMetadata, uid))

pkg/executor/executortype/poolmgr/gp.go

@@ -174,7 +174,7 @@ func (gp *GenericPool) getDeployAnnotations(env *fv1.Environment) map[string]str
 func (gp *GenericPool) checkMetricsApi() bool {
 	apiGroups, err := gp.metricsClient.DiscoveryClient.ServerGroups()
 	if err != nil {
-		gp.logger.Error("faied to discover API groups", zap.Error(err))
+		gp.logger.Error("failed to discover API groups", zap.Error(err))
 		return false
 	}
 	return utils.SupportedMetricsAPIVersionAvailable(apiGroups)
@@ -263,7 +263,13 @@ func (gp *GenericPool) choosePod(ctx context.Context, newLabels map[string]strin
 		pod, err := gp.readyPodLister.Pods(namespace).Get(name)
 		if err != nil {
 			logger.Error("fetching object from store failed", zap.String("key", key), zap.Error(err))
-			return "", nil, err
+			gp.readyPodQueue.Done(key)
+			continue
+		}
+		if utils.IsPodTerminated(pod) {
+			logger.Error("pod is terminated", zap.String("key", key))
+			gp.readyPodQueue.Done(key)
+			continue
 		}
 		if !utils.IsReadyPod(pod) {
 			logger.Warn("pod not ready, pod will be checked again", zap.String("key", key), zap.Duration("delay", expoDelay))

pkg/executor/executortype/poolmgr/gpm.go

@@ -690,7 +690,7 @@ func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes
 			if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok {
 				fsvc, ok := fsvc.(*fscache.FuncSvc)
 				if !ok {
-					gpm.logger.Error("could not covert item from PodToFsvc")
+					gpm.logger.Error("could not convert item from PodToFsvc")
 					return
 				}
 				gpm.fsCache.WebsocketFsvc.Store(fsvc.Name, true)
@@ -731,7 +731,7 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubern
 			if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok {
 				fsvc, ok := fsvc.(*fscache.FuncSvc)
 				if !ok {
-					gpm.logger.Error("could not covert value from PodToFsvc")
+					gpm.logger.Error("could not convert value from PodToFsvc")
 					return
 				}
 				ctx := context.Background()

pkg/executor/executortype/poolmgr/poolpodcontroller.go

@@ -174,7 +174,7 @@ func (p *PoolPodController) handleRSDelete(obj interface{}) {
 	if !ok {
 		tombstone, ok := obj.(k8sCache.DeletedFinalStateUnknown)
 		if !ok {
-			p.logger.Error("couldnt get object from tombstone", zap.Any("obj", obj))
+			p.logger.Error("couldn't get object from tombstone", zap.Any("obj", obj))
 			return
 		}
 		rs, ok = tombstone.Obj.(*apps.ReplicaSet)
@@ -411,7 +411,7 @@ func (p *PoolPodController) spCleanupPodQueueProcessFunc() bool {
 			p.gpm.fsCache.DeleteFunctionSvc(ctx, fsvc)
 			p.gpm.fsCache.DeleteEntry(fsvc)
 		} else {
-			p.logger.Error("could not covert item from PodToFsvc", zap.String("key", key))
+			p.logger.Error("could not convert item from PodToFsvc", zap.String("key", key))
 		}
 	}
 	err = p.kubernetesClient.CoreV1().Pods(p.namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{})

pkg/executor/fscache/functionServiceCache.go

@@ -401,7 +401,7 @@ func (fsc *FunctionServiceCache) ListOld(age time.Duration) ([]*FuncSvc, error)
 	return resp.objects, resp.error
 }
 
-// ListOldForPool returns a list of aged function serices in cache for pooling.
+// ListOldForPool returns a list of aged function services in cache for pooling.
 func (fsc *FunctionServiceCache) ListOldForPool(age time.Duration) ([]*FuncSvc, error) {
 	responseChannel := make(chan *fscResponse)
 	fsc.requestChannel <- &fscRequest{

pkg/fetcher/config/config.go

@@ -113,7 +113,7 @@ func (cfg *Config) NewSpecializeRequest(fn *fv1.Function, env *fv1.Environment)
 		if env.Spec.AllowedFunctionsPerContainer == fv1.AllowedFunctionsPerContainerInfinite {
 			// workflow loads multiple functions into one function pod,
 			// we have to use a Function UID to separate the function code
-			// to avoid overwritting.
+			// to avoid overwriting.
 			targetFilename = string(fn.ObjectMeta.UID)
 		} else {
 			// set target file name to fix pattern for

pkg/fetcher/types.go

@@ -62,7 +62,7 @@ type (
 		// to "/".
 		URL string `json:"url"`
 
-		// Metatdata
+		// Metadata
 		FunctionMetadata *metav1.ObjectMeta
 
 		EnvVersion int `json:"envVersion"`

pkg/fission-cli/flag/flag.go

@@ -45,7 +45,7 @@ type (
 		Usage        string
 		DefaultValue interface{}
 
-		// If a flag is marked as deprecated, it will hided from
+		// If a flag is marked as deprecated, it will hidden from
 		// the help message automatically. Hence, a flag cannot be
 		// marked as hidden and deprecated at the same time.
 		Hidden     bool
@@ -200,7 +200,7 @@ var (
 	SpecDelete     = Flag{Type: Bool, Name: flagkey.SpecDelete, Usage: "Allow apply to delete resources that no longer exist in the specification"}
 	SpecDry        = Flag{Type: Bool, Name: flagkey.SpecDry, Usage: "View the generated specs"}
 	SpecValidation = Flag{Type: String, Name: flagkey.SpecValidate, Usage: "Turns server side validations of Fission objects on/off"}
-	SpecIgnore     = Flag{Type: String, Name: flagkey.SpecIgnore, Usage: fmt.Sprintf("File containing specs to be ingored inside --specdir, defaults to %v", util.SPEC_IGNORE_FILE)}
+	SpecIgnore     = Flag{Type: String, Name: flagkey.SpecIgnore, Usage: fmt.Sprintf("File containing specs to be ignored inside --specdir, defaults to %v", util.SPEC_IGNORE_FILE)}
 
 	SupportOutput = Flag{Type: String, Name: flagkey.SupportOutput, Short: "o", Usage: "Output directory to save dump archive/files", DefaultValue: flagkey.DefaultSpecOutputDir}
 	SupportNoZip  = Flag{Type: Bool, Name: flagkey.SupportNoZip, Usage: "Save dump information into multiple files instead of single zip file"}

pkg/mqtrigger/messageQueue/kafka/kafka.go

@@ -17,6 +17,7 @@ limitations under the License.
 package kafka
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
@@ -28,7 +29,6 @@ import (
 	"strings"
 
 	"github.com/Shopify/sarama"
-	cluster "github.com/bsm/sarama-cluster"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 
@@ -65,6 +65,183 @@ type (
 	Factory struct{}
 )
 
+type MqtConsumerGroupHandler struct {
+	version        sarama.KafkaVersion
+	logger         *zap.Logger
+	trigger        *fv1.MessageQueueTrigger
+	fissionHeaders map[string]string
+	producer       sarama.SyncProducer
+	fnUrl          string
+}
+
+func NewMqtConsumerGroupHandler(version sarama.KafkaVersion,
+	logger *zap.Logger,
+	trigger *fv1.MessageQueueTrigger,
+	producer sarama.SyncProducer,
+	routerUrl string) MqtConsumerGroupHandler {
+	ch := MqtConsumerGroupHandler{
+		version:  version,
+		logger:   logger,
+		trigger:  trigger,
+		producer: producer,
+	}
+	// Support other function ref types
+	if ch.trigger.Spec.FunctionReference.Type != fv1.FunctionReferenceTypeFunctionName {
+		ch.logger.Fatal("unsupported function reference type for trigger",
+			zap.Any("function_reference_type", ch.trigger.Spec.FunctionReference.Type),
+			zap.String("trigger", ch.trigger.ObjectMeta.Name))
+	}
+	// Generate the Headers
+	ch.fissionHeaders = map[string]string{
+		"X-Fission-MQTrigger-Topic":      ch.trigger.Spec.Topic,
+		"X-Fission-MQTrigger-RespTopic":  ch.trigger.Spec.ResponseTopic,
+		"X-Fission-MQTrigger-ErrorTopic": ch.trigger.Spec.ErrorTopic,
+		"Content-Type":                   ch.trigger.Spec.ContentType,
+	}
+	ch.fnUrl = routerUrl + "/" + strings.TrimPrefix(utils.UrlForFunction(ch.trigger.Spec.FunctionReference.Name, ch.trigger.ObjectMeta.Namespace), "/")
+	ch.logger.Debug("function HTTP URL", zap.String("url", ch.fnUrl))
+	return ch
+}
+
+func (ch MqtConsumerGroupHandler) Setup(sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+func (ch MqtConsumerGroupHandler) Cleanup(sarama.ConsumerGroupSession) error {
+	return nil
+}
+
+func (ch MqtConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for msg := range claim.Messages() {
+		ch.kafkaMsgHandler(session, msg)
+	}
+	return nil
+}
+
+//func (ch *MqtConsumerGroupHandler) kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *fv1.MessageQueueTrigger, msg *sarama.ConsumerMessage, consumer *cluster.Consumer) {
+
+func (ch *MqtConsumerGroupHandler) kafkaMsgHandler(session sarama.ConsumerGroupSession, msg *sarama.ConsumerMessage) {
+	var value string = string(msg.Value[:])
+
+	// Create request
+	req, err := http.NewRequest("POST", ch.fnUrl, strings.NewReader(value))
+	if err != nil {
+		ch.logger.Error("failed to create HTTP request to invoke function",
+			zap.Error(err),
+			zap.String("function_url", ch.fnUrl))
+		return
+	}
+
+	// Set the headers came from Kafka record
+	// Using Header.Add() as msg.Headers may have keys with more than one value
+	if ch.version.IsAtLeast(sarama.V0_11_0_0) {
+		for _, h := range msg.Headers {
+			req.Header.Add(string(h.Key), string(h.Value))
+		}
+	} else {
+		ch.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
+			zap.Any("current_version", ch.version))
+	}
+
+	for k, v := range ch.fissionHeaders {
+		req.Header.Set(k, v)
+	}
+
+	// Make the request
+	var resp *http.Response
+	for attempt := 0; attempt <= ch.trigger.Spec.MaxRetries; attempt++ {
+		// Make the request
+		resp, err = http.DefaultClient.Do(req)
+		if err != nil {
+			ch.logger.Error("sending function invocation request failed",
+				zap.Error(err),
+				zap.String("function_url", ch.fnUrl),
+				zap.String("trigger", ch.trigger.ObjectMeta.Name))
+			continue
+		}
+		if resp == nil {
+			continue
+		}
+		if err == nil && resp.StatusCode == http.StatusOK {
+			// Success, quit retrying
+			break
+		}
+	}
+
+	generateErrorHeaders := func(errString string) []sarama.RecordHeader {
+		var errorHeaders []sarama.RecordHeader
+		if ch.version.IsAtLeast(sarama.V0_11_0_0) {
+			if count, ok := errorMessageMap[errString]; ok {
+				errorMessageMap[errString] = count + 1
+			} else {
+				errorMessageMap[errString] = 1
+			}
+			errorHeaders = append(errorHeaders, sarama.RecordHeader{Key: []byte("MessageSource"), Value: []byte(ch.trigger.Spec.Topic)})
+			errorHeaders = append(errorHeaders, sarama.RecordHeader{Key: []byte("RecycleCounter"), Value: []byte(strconv.Itoa(errorMessageMap[errString]))})
+		}
+		return errorHeaders
+	}
+
+	if resp == nil {
+		errorString := fmt.Sprintf("request exceed retries: %v", ch.trigger.Spec.MaxRetries)
+		errorHeaders := generateErrorHeaders(errorString)
+		errorHandler(ch.logger, ch.trigger, ch.producer, ch.fnUrl,
+			fmt.Errorf(errorString), errorHeaders)
+		return
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+
+	ch.logger.Debug("got response from function invocation",
+		zap.String("function_url", ch.fnUrl),
+		zap.String("trigger", ch.trigger.ObjectMeta.Name),
+		zap.String("body", string(body)))
+
+	if err != nil {
+		errorString := "request body error: " + string(body)
+		errorHeaders := generateErrorHeaders(errorString)
+		errorHandler(ch.logger, ch.trigger, ch.producer, ch.fnUrl,
+			errors.Wrapf(err, errorString), errorHeaders)
+		return
+	}
+	if resp.StatusCode != 200 {
+		errorString := fmt.Sprintf("request returned failure: %v, request body error: %v", resp.StatusCode, body)
+		errorHeaders := generateErrorHeaders(errorString)
+		errorHandler(ch.logger, ch.trigger, ch.producer, ch.fnUrl,
+			fmt.Errorf("request returned failure: %v", resp.StatusCode), errorHeaders)
+		return
+	}
+	if len(ch.trigger.Spec.ResponseTopic) > 0 {
+		// Generate Kafka record headers
+		var kafkaRecordHeaders []sarama.RecordHeader
+		if ch.version.IsAtLeast(sarama.V0_11_0_0) {
+			for k, v := range resp.Header {
+				// One key may have multiple values
+				for _, v := range v {
+					kafkaRecordHeaders = append(kafkaRecordHeaders, sarama.RecordHeader{Key: []byte(k), Value: []byte(v)})
+				}
+			}
+		} else {
+			ch.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
+				zap.Any("current_version", ch.version))
+		}
+
+		_, _, err := ch.producer.SendMessage(&sarama.ProducerMessage{
+			Topic:   ch.trigger.Spec.ResponseTopic,
+			Value:   sarama.StringEncoder(body),
+			Headers: kafkaRecordHeaders,
+		})
+		if err != nil {
+			ch.logger.Warn("failed to publish response body from function invocation to topic",
+				zap.Error(err),
+				zap.String("topic", ch.trigger.Spec.Topic),
+				zap.String("function_url", ch.fnUrl))
+			return
+		}
+	}
+	session.MarkMessage(msg, "")
+}
+
 func (factory *Factory) Create(logger *zap.Logger, mqCfg messageQueue.Config, routerUrl string) (messageQueue.MessageQueue, error) {
 	return New(logger, mqCfg, routerUrl)
 }
@@ -116,10 +293,9 @@ func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Sub
 	kafka.logger.Info("brokers set", zap.Strings("brokers", kafka.brokers))
 
 	// Create new consumer
-	consumerConfig := cluster.NewConfig()
+	consumerConfig := sarama.NewConfig()
 	consumerConfig.Consumer.Return.Errors = true
-	consumerConfig.Group.Return.Notifications = true
-	consumerConfig.Config.Version = kafka.version
+	consumerConfig.Version = kafka.version
 
 	// Create new producer
 	producerConfig := sarama.NewConfig()
@@ -142,7 +318,8 @@ func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Sub
 		consumerConfig.Net.TLS.Config = tlsConfig
 	}
 
-	consumer, err := cluster.NewConsumer(kafka.brokers, string(trigger.ObjectMeta.UID), []string{trigger.Spec.Topic}, consumerConfig)
+	consumer, err := sarama.NewConsumerGroup(kafka.brokers, string(trigger.ObjectMeta.UID), consumerConfig)
+	// consumer, err := cluster.NewConsumer(kafka.brokers, string(trigger.ObjectMeta.UID), []string{trigger.Spec.Topic}, consumerConfig)
 	kafka.logger.Info("created a new consumer", zap.Strings("brokers", kafka.brokers),
 		zap.String("input topic", trigger.Spec.Topic),
 		zap.String("output topic", trigger.Spec.ResponseTopic),
@@ -174,18 +351,14 @@ func (kafka Kafka) Subscribe(trigger *fv1.MessageQueueTrigger) (messageQueue.Sub
 		}
 	}()
 
-	// consume notifications
-	go func() {
-		for ntf := range consumer.Notifications() {
-			kafka.logger.Info("consumer notification", zap.Any("notification", ntf))
-		}
-	}()
-
+	ch := NewMqtConsumerGroupHandler(kafka.version, kafka.logger, trigger, producer, kafka.routerUrl)
 	// consume messages
 	go func() {
-		for msg := range consumer.Messages() {
-			kafka.logger.Debug("calling message handler", zap.String("message", string(msg.Value[:])))
-			go kafkaMsgHandler(&kafka, producer, trigger, msg, consumer)
+		topic := []string{trigger.Spec.Topic}
+		ctx := context.Background()
+		err = consumer.Consume(ctx, topic, ch)
+		if err != nil {
+			kafka.logger.Error("consumer error", zap.Error(err))
 		}
 	}()
 
@@ -217,146 +390,7 @@ func (kafka Kafka) getTLSConfig() (*tls.Config, error) {
 }
 
 func (kafka Kafka) Unsubscribe(subscription messageQueue.Subscription) error {
-	return subscription.(*cluster.Consumer).Close()
-}
-
-func kafkaMsgHandler(kafka *Kafka, producer sarama.SyncProducer, trigger *fv1.MessageQueueTrigger, msg *sarama.ConsumerMessage, consumer *cluster.Consumer) {
-	var value string = string(msg.Value[:])
-	// Support other function ref types
-	if trigger.Spec.FunctionReference.Type != fv1.FunctionReferenceTypeFunctionName {
-		kafka.logger.Fatal("unsupported function reference type for trigger",
-			zap.Any("function_reference_type", trigger.Spec.FunctionReference.Type),
-			zap.String("trigger", trigger.ObjectMeta.Name))
-	}
-
-	url := kafka.routerUrl + "/" + strings.TrimPrefix(utils.UrlForFunction(trigger.Spec.FunctionReference.Name, trigger.ObjectMeta.Namespace), "/")
-	kafka.logger.Debug("making HTTP request", zap.String("url", url))
-
-	// Generate the Headers
-	fissionHeaders := map[string]string{
-		"X-Fission-MQTrigger-Topic":      trigger.Spec.Topic,
-		"X-Fission-MQTrigger-RespTopic":  trigger.Spec.ResponseTopic,
-		"X-Fission-MQTrigger-ErrorTopic": trigger.Spec.ErrorTopic,
-		"Content-Type":                   trigger.Spec.ContentType,
-	}
-
-	// Create request
-	req, err := http.NewRequest("POST", url, strings.NewReader(value))
-	if err != nil {
-		kafka.logger.Error("failed to create HTTP request to invoke function",
-			zap.Error(err),
-			zap.String("function_url", url))
-		return
-	}
-
-	// Set the headers came from Kafka record
-	// Using Header.Add() as msg.Headers may have keys with more than one value
-	if kafka.version.IsAtLeast(sarama.V0_11_0_0) {
-		for _, h := range msg.Headers {
-			req.Header.Add(string(h.Key), string(h.Value))
-		}
-	} else {
-		kafka.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
-			zap.Any("current_version", kafka.version))
-	}
-
-	for k, v := range fissionHeaders {
-		req.Header.Set(k, v)
-	}
-
-	// Make the request
-	var resp *http.Response
-	for attempt := 0; attempt <= trigger.Spec.MaxRetries; attempt++ {
-		// Make the request
-		resp, err = http.DefaultClient.Do(req)
-		if err != nil {
-			kafka.logger.Error("sending function invocation request failed",
-				zap.Error(err),
-				zap.String("function_url", url),
-				zap.String("trigger", trigger.ObjectMeta.Name))
-			continue
-		}
-		if resp == nil {
-			continue
-		}
-		if err == nil && resp.StatusCode == http.StatusOK {
-			// Success, quit retrying
-			break
-		}
-	}
-
-	generateErrorHeaders := func(errString string) []sarama.RecordHeader {
-		var errorHeaders []sarama.RecordHeader
-		if kafka.version.IsAtLeast(sarama.V0_11_0_0) {
-			if count, ok := errorMessageMap[errString]; ok {
-				errorMessageMap[errString] = count + 1
-			} else {
-				errorMessageMap[errString] = 1
-			}
-			errorHeaders = append(errorHeaders, sarama.RecordHeader{Key: []byte("MessageSource"), Value: []byte(trigger.Spec.Topic)})
-			errorHeaders = append(errorHeaders, sarama.RecordHeader{Key: []byte("RecycleCounter"), Value: []byte(strconv.Itoa(errorMessageMap[errString]))})
-		}
-		return errorHeaders
-	}
-
-	if resp == nil {
-		errorString := fmt.Sprintf("request exceed retries: %v", trigger.Spec.MaxRetries)
-		errorHeaders := generateErrorHeaders(errorString)
-		errorHandler(kafka.logger, trigger, producer, url,
-			fmt.Errorf(errorString), errorHeaders)
-		return
-	}
-	defer resp.Body.Close()
-	body, err := io.ReadAll(resp.Body)
-
-	kafka.logger.Debug("got response from function invocation",
-		zap.String("function_url", url),
-		zap.String("trigger", trigger.ObjectMeta.Name),
-		zap.String("body", string(body)))
-
-	if err != nil {
-		errorString := "request body error: " + string(body)
-		errorHeaders := generateErrorHeaders(errorString)
-		errorHandler(kafka.logger, trigger, producer, url,
-			errors.Wrapf(err, errorString), errorHeaders)
-		return
-	}
-	if resp.StatusCode != 200 {
-		errorString := fmt.Sprintf("request returned failure: %v, request body error: %v", resp.StatusCode, body)
-		errorHeaders := generateErrorHeaders(errorString)
-		errorHandler(kafka.logger, trigger, producer, url,
-			fmt.Errorf("request returned failure: %v", resp.StatusCode), errorHeaders)
-		return
-	}
-	if len(trigger.Spec.ResponseTopic) > 0 {
-		// Generate Kafka record headers
-		var kafkaRecordHeaders []sarama.RecordHeader
-		if kafka.version.IsAtLeast(sarama.V0_11_0_0) {
-			for k, v := range resp.Header {
-				// One key may have multiple values
-				for _, v := range v {
-					kafkaRecordHeaders = append(kafkaRecordHeaders, sarama.RecordHeader{Key: []byte(k), Value: []byte(v)})
-				}
-			}
-		} else {
-			kafka.logger.Warn("headers are not supported by current Kafka version, needs v0.11+: no record headers to add in HTTP request",
-				zap.Any("current_version", kafka.version))
-		}
-
-		_, _, err := producer.SendMessage(&sarama.ProducerMessage{
-			Topic:   trigger.Spec.ResponseTopic,
-			Value:   sarama.StringEncoder(body),
-			Headers: kafkaRecordHeaders,
-		})
-		if err != nil {
-			kafka.logger.Warn("failed to publish response body from function invocation to topic",
-				zap.Error(err),
-				zap.String("topic", trigger.Spec.Topic),
-				zap.String("function_url", url))
-			return
-		}
-	}
-	consumer.MarkOffset(msg, "") // mark message as processed
+	return subscription.(sarama.ConsumerGroup).Close()
 }
 
 func errorHandler(logger *zap.Logger, trigger *fv1.MessageQueueTrigger, producer sarama.SyncProducer, funcUrl string, err error, errorTopicHeaders []sarama.RecordHeader) {

pkg/poolcache/poolcache.go

@@ -137,7 +137,7 @@ func (c *Cache) service() {
 					}
 					if value.activeRequests == 0 {
 						if debugLevel {
-							otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no acitve requests", zap.String("function", key1.(string)), zap.String("address", key2.(string)), zap.Int("activeRequests", value.activeRequests))
+							otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no active requests", zap.String("function", key1.(string)), zap.String("address", key2.(string)), zap.Int("activeRequests", value.activeRequests))
 						}
 						vals = append(vals, value.val)
 					}

pkg/router/functionHandler.go

@@ -152,7 +152,7 @@ func (w *fakeCloseReadCloser) RealClose() error {
 // In such a case, the RoundTripper will retry requests against the new address and give up after maxRetries.
 // However, the subsequent http call for this function will ensure the cache is invalidated.
 //
-// If GetServiceForFunction returns an error or if RoundTripper exits with an error, it get's translated into 502
+// If GetServiceForFunction returns an error or if RoundTripper exits with an error, it gets translated into 502
 // inside ServeHttp function of the reverseProxy.
 // Earlier, GetServiceForFunction was called inside handler function and fission explicitly set http status code to 500
 // if it returned an error.

pkg/storagesvc/stowClient.go

@@ -17,6 +17,7 @@ limitations under the License.
 package storagesvc
 
 import (
+	"fmt"
 	"io"
 	"mime/multipart"
 	"os"
@@ -36,7 +37,7 @@ type (
 		storage Storage
 	}
 
-	//StowClient is the wraper client for stow (Cloud storage abstraction package)
+	//StowClient is the wrapper client for stow (Cloud storage abstraction package)
 	StowClient struct {
 		logger    *zap.Logger
 		config    *storageConfig
@@ -62,6 +63,43 @@ var (
 	ErrWritingFileIntoResponse = errors.New("unable to copy item into http response")
 )
 
+func getContainer(loc stow.Location, containerName string, cursor string) (stow.Container, error) {
+	// use location.Containers to find containers that match the prefix (container name)
+	cons, cursorNew, err := loc.Containers(containerName, cursor, 1)
+	if err != nil {
+		return nil, err
+	}
+	var con stow.Container
+	for _, v := range cons {
+		c, err := loc.Container(v.ID())
+		if err != nil {
+			return nil, err
+		}
+		if c.Name() == containerName {
+			con = cons[0]
+			break
+		}
+	}
+	if con == nil && !stow.IsCursorEnd(cursorNew) {
+		_, err := getContainer(loc, containerName, cursorNew)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return con, nil
+}
+
+func getOrCreateContainer(loc stow.Location, containerName string, cursor string) (stow.Container, error) {
+	con, err := loc.CreateContainer(containerName)
+	if err != nil && (os.IsExist(err) || strings.Contains(err.Error(), "BucketAlreadyOwnedByYou")) {
+		con, err = getContainer(loc, containerName, stow.CursorStart)
+	}
+	if con == nil && err == nil {
+		err = fmt.Errorf("Storage container %s not found", containerName)
+	}
+	return con, err
+}
+
 // MakeStowClient create a new StowClient for given storage
 func MakeStowClient(logger *zap.Logger, storage Storage) (*StowClient, error) {
 	storageType := getStorageType(storage)
@@ -84,23 +122,7 @@ func MakeStowClient(logger *zap.Logger, storage Storage) (*StowClient, error) {
 	}
 	stowClient.location = loc
 
-	con, err := loc.CreateContainer(config.storage.getContainerName())
-	if err != nil && (os.IsExist(err) || strings.Contains(err.Error(), "BucketAlreadyOwnedByYou")) {
-		var cons []stow.Container
-		var cursor string
-
-		// use location.Containers to find containers that match the prefix (container name)
-		cons, cursor, err = loc.Containers(config.storage.getContainerName(), stow.CursorStart, 1)
-		if err == nil {
-			con = cons[0]
-			if !stow.IsCursorEnd(cursor) {
-				// Should only have one storage container
-				err = errors.New("Found more than one matched storage containers")
-			} else {
-				con = cons[0]
-			}
-		}
-	}
+	con, err := getOrCreateContainer(loc, config.storage.getContainerName(), stow.CursorStart)
 	if err != nil {
 		return nil, err
 	}

pkg/utils/podutils.go

@@ -48,6 +48,13 @@ func IsReadyPod(pod *v1.Pod) bool {
 	return true
 }
 
+func IsPodTerminated(pod *v1.Pod) bool {
+	if phase := pod.Status.Phase; phase != v1.PodPending && phase != v1.PodRunning && phase != v1.PodUnknown {
+		return true
+	}
+	return false
+}
+
 // PodContainerReadyStatus returns the number of ready containers and total containers present in pod
 func PodContainerReadyStatus(pod *v1.Pod) (readyContainers, noOfContainers int) {
 

pkg/utils/rbacutils.go

@@ -211,7 +211,7 @@ func RemoveSAFromRoleBindingWithRetries(ctx context.Context, logger *zap.Logger,
 
 		rbObj.Subjects = newSubjects
 
-		// cant use patch for deletes, the results become in-deterministic, so using update.
+		// can't use patch for deletes, the results become in-deterministic, so using update.
 		_, err = k8sClient.RbacV1().RoleBindings(rbObj.Namespace).Update(ctx, rbObj, metav1.UpdateOptions{})
 		switch {
 		case err == nil:

test/tests/test_obj_create_in_diff_ns.sh

@@ -218,7 +218,7 @@ pool_mgr_test_1() {
 
 main() {
     # extract the test-id generated for this CI test run, so that they can be suffixed to namespaces created as part of
-    # this test and namespaces wont clash when fission CI tests are run in parallel in the future.
+    # this test and namespaces won't clash when fission CI tests are run in parallel in the future.
     id=`echo $FISSION_NAMESPACE| cut -d"-" -f2`
 
     echo "test_id : $id"

tools/cmd-docs/main.go

@@ -41,7 +41,7 @@ func main() {
 		Short: "Generate docs for fission-cli",
 		Long:  "Generate docs for fission-cli",
 		Run: func(cmd *cobra.Command, args []string) {
-			log.Printf("Generting docs in directory %s", outdir)
+			log.Printf("Generating docs in directory %s", outdir)
 			fissionApp := app.App()
 			fissionApp.DisableAutoGenTag = true
 			fissionApp.Short = "Serverless framework for Kubernetes"

tools/crd-ref-docs/config.yaml

@@ -0,0 +1,7 @@
+processor:
+  ignoreTypes:
+    - "(CanaryConfig|Environment|Function|HTTPTrigger|KubernetesWatchTrigger|MessageQueueTrigger|Package|TimeTrigger)List$"
+  ignoreGroupVersions:
+render:
+  # Version of Kubernetes to use when generating links to Kubernetes API documentation.
+  kubernetesVersion: 1.22

tools/crd-ref-docs/header.md

@@ -0,0 +1,7 @@
+---
+title: "Fission CRD Reference"
+weight: 99
+description: >
+  Fission Custom Resources Definition(CRD) Reference
+url: /docs/crd-reference/
+---