Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Extract portforward to separate package (#728)

Browse Source 
* Extract port-forward from CLI to own package (for reuseability)
* Extract CLI logging to subpackage
This commit is contained in:
Erwin van Eyk
2018-06-13 11:26:39 +08:00
committed by Ta-Ching Chen
parent ddff7100e9
commit cf1f944069
13 changed files with 239 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fission/common.go
+5 -24
View File
@@ -37,29 +37,10 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/controller/client"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
storageSvcClient "github.com/fission/fission/storagesvc/client"
)
var (
// global verbosity of our CLI
verbosity int
)
func fatal(msg string) {
os.Stderr.WriteString(msg + "\n")
os.Exit(1)
}
func warn(msg string) {
os.Stderr.WriteString(msg + "\n")
}
func verbose(msglevel int, format string, args ...interface{}) {
if verbosity >= msglevel {
fmt.Printf(format+"\n", args...)
}
}
func getClient(serverUrl string) *client.Client {
if len(serverUrl) == 0 {
// starts local portforwarder etc.
@@ -78,7 +59,7 @@ func getClient(serverUrl string) *client.Client {
func checkErr(err error, msg string) {
if err != nil {
fatal(fmt.Sprintf("Failed to %v: %v", msg, err))
log.Fatal(fmt.Sprintf("Failed to %v: %v", msg, err))
}
}
@@ -91,7 +72,7 @@ func httpRequest(method, url, body string, headers []string) *http.Response {
method != http.MethodDelete &&
method != http.MethodPost &&
method != http.MethodPut {
fatal(fmt.Sprintf("Invalid HTTP method '%s'.", method))
log.Fatal(fmt.Sprintf("Invalid HTTP method '%s'.", method))
}
req, err := http.NewRequest(method, url, strings.NewReader(body))
@@ -171,7 +152,7 @@ func createArchive(client *client.Client, fileName string, specFile string) *fis
// make a kubernetes client
_, kubeClient, _, err := crd.GetKubernetesClient()
if err != nil {
fatal(err.Error())
log.Fatal(err.Error())
}
fissionNamespace := os.Getenv("FISSION_NAMESPACE")
@@ -179,7 +160,7 @@ func createArchive(client *client.Client, fileName string, specFile string) *fis
// get svc end point for storagesvc
service, err := kubeClient.CoreV1().Services(fissionNamespace).Get("storagesvc", metav1.GetOptions{})
if err != nil {
fatal(fmt.Sprintf("Error getting storage service object from kubernetes :%v", err.Error()))
log.Fatal(fmt.Sprintf("Error getting storage service object from kubernetes :%v", err.Error()))
}
u := strings.TrimSuffix(client.Url, "/") + "/proxy/storage"
fission/environment.go
+15 -12
View File
@@ -30,6 +30,7 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/controller/client"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
)
func getFunctionsByEnvironment(client *client.Client, envName, envNamespace string) ([]crd.Function, error) {
@@ -51,13 +52,15 @@ func envCreate(c *cli.Context) error {
envName := c.String("name")
if len(envName) == 0 {
fatal("Need a name, use --name.")
log.Fatal("Need a name, use --name.")
}
envNamespace := c.String("envNamespace")
envList, err := client.EnvironmentList(envNamespace)
if err == nil && len(envList) > 0 {
verbose(2, "%d environment(s) are present in the %s namespace. These environments are not isolated from each other; use separate namespaces if you need isolation.", len(envList), envNamespace)
log.Verbose(2, "%d environment(s) are present in the %s namespace. "+
"These environments are not isolated from each other; use separate namespaces if you need isolation.",
len(envList), envNamespace)
}
var poolsize int
@@ -69,7 +72,7 @@ func envCreate(c *cli.Context) error {
envImg := c.String("image")
if len(envImg) == 0 {
fatal("Need an image, use --image.")
log.Fatal("Need an image, use --image.")
}
envVersion := c.Int("version")
@@ -139,7 +142,7 @@ func envGet(c *cli.Context) error {
envName := c.String("name")
if len(envName) == 0 {
fatal("Need a name, use --name.")
log.Fatal("Need a name, use --name.")
}
envNamespace := c.String("envNamespace")
@@ -163,7 +166,7 @@ func envUpdate(c *cli.Context) error {
envName := c.String("name")
if len(envName) == 0 {
fatal("Need a name, use --name.")
log.Fatal("Need a name, use --name.")
}
envNamespace := c.String("envNamespace")
@@ -173,7 +176,7 @@ func envUpdate(c *cli.Context) error {
envExternalNetwork := c.Bool("externalnetwork")
if len(envImg) == 0 && len(envBuilderImg) == 0 && len(envBuildCmd) == 0 {
fatal("Need --image to specify env image, or use --builder to specify env builder, or use --buildcmd to specify new build command.")
log.Fatal("Need --image to specify env image, or use --builder to specify env builder, or use --buildcmd to specify new build command.")
}
env, err := client.EnvironmentGet(&metav1.ObjectMeta{
@@ -187,7 +190,7 @@ func envUpdate(c *cli.Context) error {
}
if env.Spec.Version == 1 && (len(envBuilderImg) > 0 || len(envBuildCmd) > 0) {
fatal("Version 1 Environments do not support builders. Must specify --version=2.")
log.Fatal("Version 1 Environments do not support builders. Must specify --version=2.")
}
if len(envBuilderImg) > 0 {
@@ -219,7 +222,7 @@ func envDelete(c *cli.Context) error {
envName := c.String("name")
if len(envName) == 0 {
fatal("Need a name , use --name.")
log.Fatal("Need a name , use --name.")
}
envNamespace := c.String("envNamespace")
@@ -269,7 +272,7 @@ func getResourceReq(c *cli.Context, resources v1.ResourceRequirements) v1.Resour
mincpu := c.Int("mincpu")
cpuRequest, err := resource.ParseQuantity(strconv.Itoa(mincpu) + "m")
if err != nil {
fatal("Failed to parse mincpu")
log.Fatal("Failed to parse mincpu")
}
requestResources[v1.ResourceCPU] = cpuRequest
}
@@ -278,7 +281,7 @@ func getResourceReq(c *cli.Context, resources v1.ResourceRequirements) v1.Resour
minmem := c.Int("minmemory")
memRequest, err := resource.ParseQuantity(strconv.Itoa(minmem) + "Mi")
if err != nil {
fatal("Failed to parse minmemory")
log.Fatal("Failed to parse minmemory")
}
requestResources[v1.ResourceMemory] = memRequest
}
@@ -294,7 +297,7 @@ func getResourceReq(c *cli.Context, resources v1.ResourceRequirements) v1.Resour
maxcpu := c.Int("maxcpu")
cpuLimit, err := resource.ParseQuantity(strconv.Itoa(maxcpu) + "m")
if err != nil {
fatal("Failed to parse maxcpu")
log.Fatal("Failed to parse maxcpu")
}
limitResources[v1.ResourceCPU] = cpuLimit
}
@@ -303,7 +306,7 @@ func getResourceReq(c *cli.Context, resources v1.ResourceRequirements) v1.Resour
maxmem := c.Int("maxmemory")
memLimit, err := resource.ParseQuantity(strconv.Itoa(maxmem) + "Mi")
if err != nil {
fatal("Failed to parse maxmemory")
log.Fatal("Failed to parse maxmemory")
}
limitResources[v1.ResourceMemory] = memLimit
}
fission/function.go
+35 -33
View File
@@ -36,13 +36,15 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
"github.com/fission/fission/fission/logdb"
"github.com/fission/fission/fission/portforward"
)
func printPodLogs(c *cli.Context) error {
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need --name argument.")
log.Fatal("Need --name argument.")
}
queryURL, err := url.Parse(getServerUrl())
@@ -74,7 +76,7 @@ func getInvokeStrategy(minScale int, maxScale int, executorType string, targetcp
}
if minScale > maxScale {
fatal("Maxscale must be higher than or equal to minscale")
log.Fatal("Maxscale must be higher than or equal to minscale")
}
var fnExecutor fission.ExecutorType
@@ -86,7 +88,7 @@ func getInvokeStrategy(minScale int, maxScale int, executorType string, targetcp
case fission.ExecutorTypeNewdeploy:
fnExecutor = fission.ExecutorTypeNewdeploy
default:
fatal("Executor type must be one of 'poolmgr' or 'newdeploy', defaults to 'poolmgr'")
log.Fatal("Executor type must be one of 'poolmgr' or 'newdeploy', defaults to 'poolmgr'")
}
// Right now a simple single case strategy implementation
@@ -108,7 +110,7 @@ func getTargetCPU(c *cli.Context) int {
if c.IsSet("targetcpu") {
targetCPU = c.Int("targetcpu")
if targetCPU <= 0 || targetCPU > 100 {
fatal("TargetCPU must be a value between 1 - 100")
log.Fatal("TargetCPU must be a value between 1 - 100")
}
} else {
targetCPU = 80
@@ -125,7 +127,7 @@ func fnCreate(c *cli.Context) error {
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need --name argument.")
log.Fatal("Need --name argument.")
}
// user wants a spec, create a yaml file with package and function
@@ -142,7 +144,7 @@ func fnCreate(c *cli.Context) error {
// check function existence before creating package
for _, fn := range fnList {
if fn.Metadata.Name == fnName {
fatal("A function with the same name already exists.")
log.Fatal("A function with the same name already exists.")
}
}
entrypoint := c.String("entrypoint")
@@ -168,7 +170,7 @@ func fnCreate(c *cli.Context) error {
// need to specify environment for creating new package
envName = c.String("env")
if len(envName) == 0 {
fatal("Need --env argument.")
log.Fatal("Need --env argument.")
}
// examine existence of given environment
@@ -191,7 +193,7 @@ func fnCreate(c *cli.Context) error {
}
// fatal when both src & deploy archive are empty
if len(srcArchiveName) == 0 && len(deployArchiveName) == 0 {
fatal("Need --deploy or --src argument.")
log.Fatal("Need --deploy or --src argument.")
}
buildcmd := c.String("buildcmd")
@@ -204,7 +206,7 @@ func fnCreate(c *cli.Context) error {
resourceReq := getResourceReq(c, apiv1.ResourceRequirements{})
if (c.IsSet("mincpu") || c.IsSet("maxcpu") || c.IsSet("minmemory") || c.IsSet("maxmemory")) &&
invokeStrategy.ExecutionStrategy.ExecutorType == fission.ExecutorTypePoolmgr {
warn("CPU/Memory specified for function with pool manager executor will be ignored in favor of resources specified at environment")
log.Warn("CPU/Memory specified for function with pool manager executor will be ignored in favor of resources specified at environment")
}
var secrets []fission.SecretReference
@@ -217,7 +219,7 @@ func fnCreate(c *cli.Context) error {
Name: secretName,
})
if k8serrors.IsNotFound(err) {
warn(fmt.Sprintf("Secret %s not found in Namespace: %s. Secret needs to be present in the same namespace as function", secretName, fnNamespace))
log.Warn(fmt.Sprintf("Secret %s not found in Namespace: %s. Secret needs to be present in the same namespace as function", secretName, fnNamespace))
}
newSecret := fission.SecretReference{
@@ -234,7 +236,7 @@ func fnCreate(c *cli.Context) error {
Name: cfgMapName,
})
if k8serrors.IsNotFound(err) {
warn(fmt.Sprintf("ConfigMap %s not found in Namespace: %s. ConfigMap needs to be present in the same namespace as function", cfgMapName, fnNamespace))
log.Warn(fmt.Sprintf("ConfigMap %s not found in Namespace: %s. ConfigMap needs to be present in the same namespace as function", cfgMapName, fnNamespace))
}
newCfgMap := fission.ConfigMapReference{
@@ -322,7 +324,7 @@ func fnGet(c *cli.Context) error {
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need name of function, use --name")
log.Fatal("Need name of function, use --name")
}
fnNamespace := c.String("fnNamespace")
m := &metav1.ObjectMeta{
@@ -347,7 +349,7 @@ func fnGetMeta(c *cli.Context) error {
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need name of function, use --name")
log.Fatal("Need name of function, use --name")
}
fnNamespace := c.String("fnNamespace")
@@ -371,16 +373,16 @@ func fnUpdate(c *cli.Context) error {
client := getClient(c.GlobalString("server"))
if len(c.String("package")) > 0 {
fatal("--package is deprecated, please use --deploy instead.")
log.Fatal("--package is deprecated, please use --deploy instead.")
}
if len(c.String("srcpkg")) > 0 {
fatal("--srcpkg is deprecated, please use --src instead.")
log.Fatal("--srcpkg is deprecated, please use --src instead.")
}
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need name of function, use --name")
log.Fatal("Need name of function, use --name")
}
fnNamespace := c.String("fnNamespace")
@@ -406,12 +408,12 @@ func fnUpdate(c *cli.Context) error {
cfgMapName := c.String("configmap")
if len(srcArchiveName) > 0 && len(deployArchiveName) > 0 {
fatal("Need either of --src or --deploy and not both arguments.")
log.Fatal("Need either of --src or --deploy and not both arguments.")
}
if len(secretName) > 0 {
if len(function.Spec.Secrets) > 1 {
fatal("Please use 'fission spec apply' to update list of secrets")
log.Fatal("Please use 'fission spec apply' to update list of secrets")
}
// check that the referenced secret is in the same ns as the function, if not give a warning.
@@ -420,7 +422,7 @@ func fnUpdate(c *cli.Context) error {
Name: secretName,
})
if k8serrors.IsNotFound(err) {
warn(fmt.Sprintf("secret %s not found in Namespace: %s. Secret needs to be present in the same namespace as function", secretName, fnNamespace))
log.Warn(fmt.Sprintf("secret %s not found in Namespace: %s. Secret needs to be present in the same namespace as function", secretName, fnNamespace))
}
newSecret := fission.SecretReference{
@@ -432,7 +434,7 @@ func fnUpdate(c *cli.Context) error {
if len(cfgMapName) > 0 {
if len(function.Spec.ConfigMaps) > 1 {
fatal("Please use 'fission spec apply' to update list of configmaps")
log.Fatal("Please use 'fission spec apply' to update list of configmaps")
}
// check that the referenced cfgmap is in the same ns as the function, if not give a warning.
@@ -441,7 +443,7 @@ func fnUpdate(c *cli.Context) error {
Name: cfgMapName,
})
if k8serrors.IsNotFound(err) {
warn(fmt.Sprintf("ConfigMap %s not found in Namespace: %s. ConfigMap needs to be present in the same namespace as the function", cfgMapName, fnNamespace))
log.Warn(fmt.Sprintf("ConfigMap %s not found in Namespace: %s. ConfigMap needs to be present in the same namespace as the function", cfgMapName, fnNamespace))
}
newCfgMap := fission.ConfigMapReference{
@@ -476,7 +478,7 @@ func fnUpdate(c *cli.Context) error {
checkErr(err, "get function list")
if !force && len(fnList) > 1 {
fatal("Package is used by multiple functions, use --force to force update")
log.Fatal("Package is used by multiple functions, use --force to force update")
}
pkgMetadata = updatePackage(client, pkg, envName, envNamespace, srcArchiveName, deployArchiveName, buildcmd)
@@ -515,11 +517,11 @@ func fnUpdate(c *cli.Context) error {
minscale := c.Int("minscale")
maxscale := c.Int("maxscale")
if c.IsSet("maxscale") && minscale > c.Int("maxscale") {
fatal(fmt.Sprintf("Minscale's value %v can not be greater than maxscale value %v", minscale, maxscale))
log.Fatal(fmt.Sprintf("Minscale's value %v can not be greater than maxscale value %v", minscale, maxscale))
}
if function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType != fission.ExecutorTypePoolmgr &&
minscale > function.Spec.InvokeStrategy.ExecutionStrategy.MaxScale {
fatal(fmt.Sprintf("Minscale provided: %v can not be greater than maxscale of existing function: %v", minscale,
log.Fatal(fmt.Sprintf("Minscale provided: %v can not be greater than maxscale of existing function: %v", minscale,
function.Spec.InvokeStrategy.ExecutionStrategy.MaxScale))
}
function.Spec.InvokeStrategy.ExecutionStrategy.MinScale = minscale
@@ -528,7 +530,7 @@ func fnUpdate(c *cli.Context) error {
if c.IsSet("maxscale") {
maxscale := c.Int("maxscale")
if maxscale < function.Spec.InvokeStrategy.ExecutionStrategy.MinScale {
fatal(fmt.Sprintf("Function's minscale: %v can not be greater than maxscale provided: %v",
log.Fatal(fmt.Sprintf("Function's minscale: %v can not be greater than maxscale provided: %v",
function.Spec.InvokeStrategy.ExecutionStrategy.MinScale, maxscale))
}
function.Spec.InvokeStrategy.ExecutionStrategy.MaxScale = maxscale
@@ -544,11 +546,11 @@ func fnUpdate(c *cli.Context) error {
case fission.ExecutorTypeNewdeploy:
fnExecutor = fission.ExecutorTypeNewdeploy
default:
fatal("Executor type must be one of 'poolmgr' or 'newdeploy', defaults to 'poolmgr'")
log.Fatal("Executor type must be one of 'poolmgr' or 'newdeploy', defaults to 'poolmgr'")
}
if (c.IsSet("mincpu") || c.IsSet("maxcpu") || c.IsSet("minmemory") || c.IsSet("maxmemory")) &&
fnExecutor == fission.ExecutorTypePoolmgr {
warn("CPU/Memory specified for function with pool manager executor will be ignored in favor of resources specified at environment")
log.Warn("CPU/Memory specified for function with pool manager executor will be ignored in favor of resources specified at environment")
}
function.Spec.InvokeStrategy.ExecutionStrategy.ExecutorType = fnExecutor
}
@@ -565,7 +567,7 @@ func fnDelete(c *cli.Context) error {
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need name of function, use --name")
log.Fatal("Need name of function, use --name")
}
fnNamespace := c.String("fnNamespace")
@@ -616,7 +618,7 @@ func fnLogs(c *cli.Context) error {
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need name of function, use --name")
log.Fatal("Need name of function, use --name")
}
fnNamespace := c.String("fnNamespace")
@@ -642,7 +644,7 @@ func fnLogs(c *cli.Context) error {
// request the controller to establish a proxy server to the database.
logDB, err := logdb.GetLogDB(dbType, getServerUrl())
if err != nil {
fatal("failed to connect log database")
log.Fatal("failed to connect log database")
}
requestChan := make(chan struct{})
@@ -663,7 +665,7 @@ func fnLogs(c *cli.Context) error {
}
logEntries, err := logDB.GetLogs(logFilter)
if err != nil {
fatal("failed to query logs")
log.Fatal("failed to query logs")
}
for _, logEntry := range logEntries {
if c.Bool("d") {
@@ -695,14 +697,14 @@ func fnLogs(c *cli.Context) error {
func fnTest(c *cli.Context) error {
fnName := c.String("name")
if len(fnName) == 0 {
fatal("Need function name to be specified with --name")
log.Fatal("Need function name to be specified with --name")
}
ns := c.String("fnNamespace")
routerURL := os.Getenv("FISSION_ROUTER")
if len(routerURL) == 0 {
// Portforward to the fission router
localRouterPort := setupPortForward(getKubeConfigPath(),
localRouterPort := portforward.Setup(getKubeConfigPath(),
getFissionNamespace(), "application=fission-router")
routerURL = "127.0.0.1:" + localRouterPort
} else {
fission/httptrigger.go
+7 -6
View File
@@ -30,6 +30,7 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/controller/client"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
)
// returns one of http.Method*
@@ -54,7 +55,7 @@ func getMethod(method string) string {
case "TRACE":
return http.MethodTrace
}
fatal(fmt.Sprintf("Invalid HTTP Method %v", method))
log.Fatal(fmt.Sprintf("Invalid HTTP Method %v", method))
return ""
}
@@ -75,13 +76,13 @@ func htCreate(c *cli.Context) error {
fnName := c.String("function")
if len(fnName) == 0 {
fatal("Need a function name to create a trigger, use --function")
log.Fatal("Need a function name to create a trigger, use --function")
}
fnNamespace := c.String("fnNamespace")
triggerUrl := c.String("url")
if len(triggerUrl) == 0 {
fatal("Need a trigger URL, use --url")
log.Fatal("Need a trigger URL, use --url")
}
if !strings.HasPrefix(triggerUrl, "/") {
triggerUrl = fmt.Sprintf("/%s", triggerUrl)
@@ -143,14 +144,14 @@ func htUpdate(c *cli.Context) error {
client := getClient(c.GlobalString("server"))
htName := c.String("name")
if len(htName) == 0 {
fatal("Need name of trigger, use --name")
log.Fatal("Need name of trigger, use --name")
}
triggerNamespace := c.String("triggerNamespace")
// update function ref
newFn := c.String("function")
if len(newFn) == 0 {
fatal("Nothing to update. Use --function to specify a new function.")
log.Fatal("Nothing to update. Use --function to specify a new function.")
}
checkFunctionExistence(client, newFn, triggerNamespace)
@@ -184,7 +185,7 @@ func htDelete(c *cli.Context) error {
client := getClient(c.GlobalString("server"))
htName := c.String("name")
if len(htName) == 0 {
fatal("Need name of trigger to delete, use --name")
log.Fatal("Need name of trigger to delete, use --name")
}
triggerNamespace := c.String("triggerNamespace")
fission/log/log.go
+42
View File
@@ -0,0 +1,42 @@
/*
Copyright 2016 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 log
import (
"fmt"
"os"
)
var (
// global Verbosity of our CLI
Verbosity int
)
func Fatal(msg interface{}) {
os.Stderr.WriteString(fmt.Sprintf("%v\n", msg))
os.Exit(1)
}
func Warn(msg interface{}) {
os.Stderr.WriteString(fmt.Sprintf("%v\n", msg))
}
func Verbose(verbosityLevel int, format string, args ...interface{}) {
if Verbosity >= verbosityLevel {
fmt.Printf(format+"\n", args...)
}
}
fission/main.go
+7 -5
View File
@@ -22,6 +22,8 @@ import (
"path/filepath"
"github.com/fission/fission"
"github.com/fission/fission/fission/log"
"github.com/fission/fission/fission/portforward"
"github.com/urfave/cli"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
@@ -38,7 +40,8 @@ func getKubeConfigPath() string {
kubeConfig = filepath.Join(home, ".kube", "config")
if _, err := os.Stat(kubeConfig); os.IsNotExist(err) {
fatal("Couldn't find kubeconfig file. Set the KUBECONFIG environment variable to your kubeconfig's path.")
log.Fatal("Couldn't find kubeconfig file. " +
"Set the KUBECONFIG environment variable to your kubeconfig's path.")
}
}
return kubeConfig
@@ -51,8 +54,7 @@ func getServerUrl() string {
if len(fissionUrl) == 0 {
fissionNamespace := getFissionNamespace()
kubeConfig := getKubeConfigPath()
localPort := setupPortForward(
kubeConfig, fissionNamespace, "application=fission-api")
localPort := portforward.Setup(kubeConfig, fissionNamespace, "application=fission-api")
serverUrl = "http://127.0.0.1:" + localPort
} else {
serverUrl = fissionUrl
@@ -61,8 +63,8 @@ func getServerUrl() string {
}
func cliHook(c *cli.Context) error {
verbosity = c.Int("verbosity")
verbose(2, "Verbosity = 2")
log.Verbosity = c.Int("verbosity")
log.Verbose(2, "Verbosity = 2")
return nil
}
fission/mqtrigger.go
+9 -8
View File
@@ -27,6 +27,7 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
fv1 "github.com/fission/fission/pkg/apis/fission.io/v1"
)
@@ -39,7 +40,7 @@ func mqtCreate(c *cli.Context) error {
}
fnName := c.String("function")
if len(fnName) == 0 {
fatal("Need a function name to create a trigger, use --function")
log.Fatal("Need a function name to create a trigger, use --function")
}
fnNamespace := c.String("fnNamespace")
@@ -52,20 +53,20 @@ func mqtCreate(c *cli.Context) error {
case fission.MessageQueueTypeASQ:
mqType = fission.MessageQueueTypeASQ
default:
fatal("Unknown message queue type, currently only \"nats-streaming, azure-storage-queue \" is supported")
log.Fatal("Unknown message queue type, currently only \"nats-streaming, azure-storage-queue \" is supported")
}
// TODO: check topic availability
topic := c.String("topic")
if len(topic) == 0 {
fatal("Listen topic cannot be empty")
log.Fatal("Listen topic cannot be empty")
}
respTopic := c.String("resptopic")
if topic == respTopic {
// TODO maybe this should just be a warning, perhaps
// allow it behind a --force flag
fatal("Listen topic should not equal to response topic")
log.Fatal("Listen topic should not equal to response topic")
}
contentType := c.String("contenttype")
@@ -115,7 +116,7 @@ func mqtUpdate(c *cli.Context) error {
client := getClient(c.GlobalString("server"))
mqtName := c.String("name")
if len(mqtName) == 0 {
fatal("Need name of trigger, use --name")
log.Fatal("Need name of trigger, use --name")
}
mqtNs := c.String("triggerns")
@@ -153,7 +154,7 @@ func mqtUpdate(c *cli.Context) error {
}
if !updated {
fatal("Nothing to update. Use --topic, --resptopic, or --function.")
log.Fatal("Nothing to update. Use --topic, --resptopic, or --function.")
}
_, err = client.MessageQueueTriggerUpdate(mqt)
@@ -167,7 +168,7 @@ func mqtDelete(c *cli.Context) error {
client := getClient(c.GlobalString("server"))
mqtName := c.String("name")
if len(mqtName) == 0 {
fatal("Need name of trigger to delete, use --name")
log.Fatal("Need name of trigger to delete, use --name")
}
mqtNs := c.String("triggerns")
@@ -204,7 +205,7 @@ func mqtList(c *cli.Context) error {
func checkMQTopicAvailability(mqType fission.MessageQueueType, topics ...string) {
for _, t := range topics {
if len(t) > 0 && !fv1.IsTopicValid(mqType, t) {
fatal(fmt.Sprintf("Invalid topic for %s: %s", mqType, t))
log.Fatal(fmt.Sprintf("Invalid topic for %s: %s", mqType, t))
}
}
}
fission/package.go
+11 -10
View File
@@ -31,6 +31,7 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/controller/client"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
)
func getFunctionsByPackage(client *client.Client, pkgName, pkgNamespace string) ([]crd.Function, error) {
@@ -68,7 +69,7 @@ func pkgCreate(c *cli.Context) error {
pkgNamespace := c.String("pkgNamespace")
envName := c.String("env")
if len(envName) == 0 {
fatal("Need --env argument.")
log.Fatal("Need --env argument.")
}
envNamespace := c.String("envNamespace")
srcArchiveName := c.String("src")
@@ -76,7 +77,7 @@ func pkgCreate(c *cli.Context) error {
buildcmd := c.String("buildcmd")
if len(srcArchiveName) == 0 && len(deployArchiveName) == 0 {
fatal("Need --src to specify source archive, or use --deploy to specify deployment archive.")
log.Fatal("Need --src to specify source archive, or use --deploy to specify deployment archive.")
}
meta := createPackage(client, pkgNamespace, envName, envNamespace, srcArchiveName, deployArchiveName, buildcmd, "")
@@ -90,7 +91,7 @@ func pkgUpdate(c *cli.Context) error {
pkgName := c.String("name")
if len(pkgName) == 0 {
fatal("Need --name argument.")
log.Fatal("Need --name argument.")
}
pkgNamespace := c.String("pkgNamespace")
@@ -102,12 +103,12 @@ func pkgUpdate(c *cli.Context) error {
buildcmd := c.String("buildcmd")
if len(srcArchiveName) > 0 && len(deployArchiveName) > 0 {
fatal("Need either of --src or --deploy and not both arguments.")
log.Fatal("Need either of --src or --deploy and not both arguments.")
}
if len(srcArchiveName) == 0 && len(deployArchiveName) == 0 &&
len(envName) == 0 && len(buildcmd) == 0 {
fatal("Need --env or --src or --deploy or --buildcmd argument.")
log.Fatal("Need --env or --src or --deploy or --buildcmd argument.")
}
pkg, err := client.PackageGet(&metav1.ObjectMeta{
@@ -120,7 +121,7 @@ func pkgUpdate(c *cli.Context) error {
checkErr(err, "get function list")
if !force && len(fnList) > 1 {
fatal("Package is used by multiple functions, use --force to force update")
log.Fatal("Package is used by multiple functions, use --force to force update")
}
newPkgMeta := updatePackage(client, pkg,
@@ -185,7 +186,7 @@ func pkgSourceGet(c *cli.Context) error {
pkgName := c.String("name")
if len(pkgName) == 0 {
fatal("Need name of package, use --name")
log.Fatal("Need name of package, use --name")
}
pkgNamespace := c.String("pkgNamespace")
@@ -222,7 +223,7 @@ func pkgDeployGet(c *cli.Context) error {
pkgName := c.String("name")
if len(pkgName) == 0 {
fatal("Need name of package, use --name")
log.Fatal("Need name of package, use --name")
}
pkgNamespace := c.String("pkgNamespace")
@@ -259,7 +260,7 @@ func pkgInfo(c *cli.Context) error {
pkgName := c.String("name")
if len(pkgName) == 0 {
fatal("Need name of package, use --name")
log.Fatal("Need name of package, use --name")
}
pkgNamespace := c.String("pkgNamespace")
@@ -369,7 +370,7 @@ func pkgDelete(c *cli.Context) error {
fnList, err := getFunctionsByPackage(client, pkgName, pkgNamespace)
if !force && len(fnList) > 0 {
fatal("Package is used by at least one function, use -f to force delete")
log.Fatal("Package is used by at least one function, use -f to force delete")
}
err = deletePackage(client, pkgName, pkgNamespace)
fission/portforward.go → fission/portforward/portforward.go
+86 -68
View File
@@ -1,4 +1,20 @@
package main
/*
Copyright 2016 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 portforward
import (
"fmt"
@@ -14,8 +30,60 @@ import (
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/tools/portforward"
"k8s.io/client-go/transport/spdy"
"github.com/fission/fission/fission/log"
)
// Port forward a free local port to a pod on the cluster. The pod is
// found in the specified namespace by labelSelector. The pod's port
// is found by looking for a service in the same namespace and using
// its targetPort. Once the port forward is started, wait for it to
// start accepting connections before returning.
func Setup(kubeConfig, namespace, labelSelector string) string {
log.Verbose(2, "Setting up port forward to %s in namespace %s using the kubeconfig at %s",
labelSelector, namespace, kubeConfig)
localPort, err := findFreePort()
if err != nil {
log.Fatal(fmt.Sprintf("Error finding unused port :%v", err.Error()))
}
log.Verbose(2, "Waiting for local port %v", localPort)
for {
conn, _ := net.DialTimeout("tcp",
net.JoinHostPort("", localPort), time.Millisecond)
if conn != nil {
conn.Close()
} else {
break
}
time.Sleep(time.Millisecond * 50)
}
log.Verbose(2, "Starting port forward from local port %v", localPort)
go func() {
err := runPortForward(kubeConfig, labelSelector, localPort, namespace)
if err != nil {
log.Fatal(fmt.Sprintf("Error forwarding to controller port: %s", err.Error()))
}
}()
log.Verbose(2, "Waiting for port forward %v to start...", localPort)
for {
conn, _ := net.DialTimeout("tcp",
net.JoinHostPort("", localPort), time.Millisecond)
if conn != nil {
conn.Close()
break
}
time.Sleep(time.Millisecond * 50)
}
log.Verbose(2, "Port forward from local port %v started", localPort)
return localPort
}
func findFreePort() (string, error) {
listener, err := net.Listen("tcp", ":0")
if err != nil {
@@ -33,29 +101,29 @@ func findFreePort() (string, error) {
}
// runPortForward creates a local port forward to the specified pod
func runPortForward(kubeConfig string, labelSelector string, localPort string, fissionNamespace string) error {
func runPortForward(kubeConfig string, labelSelector string, localPort string, ns string) error {
config, err := clientcmd.BuildConfigFromFlags("", kubeConfig)
if err != nil {
fatal(fmt.Sprintf("Failed to connect to Kubernetes: %s", err))
log.Fatal(fmt.Sprintf("Failed to connect to Kubernetes: %s", err))
}
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
fatal(fmt.Sprintf("Failed to connect to Kubernetes: %s", err))
log.Fatal(fmt.Sprintf("Failed to connect to Kubernetes: %s", err))
}
verbose(2, "Connected to Kubernetes API")
log.Verbose(2, "Connected to Kubernetes API")
// if fission namespace is unset, try to find a fission pod in any namespace
if len(fissionNamespace) == 0 {
fissionNamespace = meta_v1.NamespaceAll
// if namespace is unset, try to find a pod in any namespace
if len(ns) == 0 {
ns = meta_v1.NamespaceAll
}
// get the pod; if there is more than one, ask the user to disambiguate
podList, err := clientset.CoreV1().Pods(fissionNamespace).
podList, err := clientset.CoreV1().Pods(ns).
List(meta_v1.ListOptions{LabelSelector: labelSelector})
if err != nil || len(podList.Items) == 0 {
fatal("Error getting controller pod for port-forwarding")
log.Fatal("Error getting controller pod for port-forwarding")
}
// make a useful error message if there is more than one install
@@ -64,7 +132,7 @@ func runPortForward(kubeConfig string, labelSelector string, localPort string, f
for _, p := range podList.Items {
namespaces = append(namespaces, p.Namespace)
}
fatal(fmt.Sprintf("Found %v fission installs, set FISSION_NAMESPACE to one of: %v",
log.Fatal(fmt.Sprintf("Found %v fission installs, set FISSION_NAMESPACE to one of: %v",
len(podList.Items), strings.Join(namespaces, " ")))
}
@@ -76,10 +144,10 @@ func runPortForward(kubeConfig string, labelSelector string, localPort string, f
svcs, err := clientset.CoreV1().Services(podNameSpace).
List(meta_v1.ListOptions{LabelSelector: labelSelector})
if err != nil {
fatal(fmt.Sprintf("Error getting %v service :%v", labelSelector, err.Error()))
log.Fatal(fmt.Sprintf("Error getting %v service :%v", labelSelector, err.Error()))
}
if len(svcs.Items) == 0 {
fatal(fmt.Sprintf("Service %v not found", labelSelector))
log.Fatal(fmt.Sprintf("Service %v not found", labelSelector))
}
service := &svcs.Items[0]
@@ -87,7 +155,7 @@ func runPortForward(kubeConfig string, labelSelector string, localPort string, f
for _, servicePort := range service.Spec.Ports {
targetPort = servicePort.TargetPort.String()
}
verbose(2, "Connecting to port %v on pod %v/%v", targetPort, podNameSpace, podNameSpace)
log.Verbose(2, "Connecting to port %v on pod %v/%v", targetPort, podNameSpace, podNameSpace)
stopChannel := make(chan struct{}, 1)
readyChannel := make(chan struct{})
@@ -105,70 +173,20 @@ func runPortForward(kubeConfig string, labelSelector string, localPort string, f
transport, upgrader, err := spdy.RoundTripperFor(config)
if err != nil {
msg := fmt.Sprintf("Failed to connect to Fission service on Kubernetes: %v", err.Error())
fatal(msg)
log.Fatal(msg)
}
dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", url)
outStream := os.Stdout
if verbosity < 2 {
if log.Verbosity < 2 {
outStream = nil
}
fw, err := portforward.New(dialer, ports, stopChannel, readyChannel, outStream, os.Stderr)
if err != nil {
msg := fmt.Sprintf("portforward.new errored out :%v", err.Error())
fatal(msg)
log.Fatal(msg)
}
verbose(2, "Starting port forwarder")
log.Verbose(2, "Starting port forwarder")
return fw.ForwardPorts()
}
// Port forward a free local port to a pod on the cluster. The pod is
// found in the specified namespace by labelSelector. The pod's port
// is found by looking for a service in the same namespace and using
// its targetPort. Once the port forward is started, wait for it to
// start accepting connections before returning.
func setupPortForward(kubeConfig, namespace, labelSelector string) string {
verbose(2, "Setting up port forward to %s in namespace %s using the kubeconfig at %s",
labelSelector, namespace, kubeConfig)
localPort, err := findFreePort()
if err != nil {
fatal(fmt.Sprintf("Error finding unused port :%v", err.Error()))
}
verbose(2, "Waiting for local port %v", localPort)
for {
conn, _ := net.DialTimeout("tcp",
net.JoinHostPort("", localPort), time.Millisecond)
if conn != nil {
conn.Close()
} else {
break
}
time.Sleep(time.Millisecond * 50)
}
verbose(2, "Starting port forward from local port %v", localPort)
go func() {
err := runPortForward(kubeConfig, labelSelector, localPort, namespace)
if err != nil {
fatal(fmt.Sprintf("Error forwarding to controller port: %s", err.Error()))
}
}()
verbose(2, "Waiting for port forward %v to start...", localPort)
for {
conn, _ := net.DialTimeout("tcp",
net.JoinHostPort("", localPort), time.Millisecond)
if conn != nil {
conn.Close()
break
}
time.Sleep(time.Millisecond * 50)
}
verbose(2, "Port forward from local port %v started", localPort)
return localPort
}
fission/spec.go
+5 -3
View File
@@ -39,6 +39,7 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/controller/client"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
)
const SPEC_API_VERSION = "fission.io/v1"
@@ -503,7 +504,7 @@ func (fr *FissionResources) parseYaml(b []byte, loc *location) error {
default:
// no need to error out just because there's some extra files around;
// also good for compatibility.
warn(fmt.Sprintf("Ignoring unknown type %v in %v", tm.Kind, loc))
log.Warn(fmt.Sprintf("Ignoring unknown type %v in %v", tm.Kind, loc))
}
// add to source map, check for duplicates
@@ -523,7 +524,8 @@ func readSpecs(specDir string) (*FissionResources, error) {
// make sure spec directory exists before continue
if _, err := os.Stat(specDir); os.IsNotExist(err) {
fatal(fmt.Sprintf("Spec directory %v doesn't exist. Please check directory path or run \"fission spec init\" to create it.", specDir))
log.Fatal(fmt.Sprintf("Spec directory %v doesn't exist. "+
"Please check directory path or run \"fission spec init\" to create it.", specDir))
}
fr := FissionResources{
@@ -950,7 +952,7 @@ func localArchiveFromSpec(specDir string, aus *ArchiveUploadSpec) (*fission.Arch
absGlob := rootDir + "/" + relativeGlob
f, err := filepath.Glob(absGlob)
if err != nil {
warn(fmt.Sprintf("Invalid glob in archive %v: %v", aus.Name, relativeGlob))
log.Warn(fmt.Sprintf("Invalid glob in archive %v: %v", aus.Name, relativeGlob))
return nil, err
}
files = append(files, f...)
fission/timetrigger.go
+8 -7
View File
@@ -30,12 +30,13 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/controller/client"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
)
func getAPITimeInfo(client *client.Client) time.Time {
serverInfo, err := client.ServerInfo()
if err != nil {
fatal(fmt.Sprintf("Error syncing server time information: %v", err))
log.Fatal(fmt.Sprintf("Error syncing server time information: %v", err))
}
return serverInfo.ServerTime.CurrentTime
}
@@ -65,14 +66,14 @@ func ttCreate(c *cli.Context) error {
}
fnName := c.String("function")
if len(fnName) == 0 {
fatal("Need a function name to create a trigger, use --function")
log.Fatal("Need a function name to create a trigger, use --function")
}
fnNamespace := c.String("fnNamespace")
cronSpec := c.String("cron")
if len(cronSpec) == 0 {
fatal("Need a cron spec like '0 30 * * * *', '@every 1h30m', or '@hourly'; use --cron")
log.Fatal("Need a cron spec like '0 30 * * * *', '@every 1h30m', or '@hourly'; use --cron")
}
tt := &crd.TimeTrigger{
@@ -116,7 +117,7 @@ func ttUpdate(c *cli.Context) error {
client := getClient(c.GlobalString("server"))
ttName := c.String("name")
if len(ttName) == 0 {
fatal("Need name of trigger, use --name")
log.Fatal("Need name of trigger, use --name")
}
ttNs := c.String("triggerns")
@@ -143,7 +144,7 @@ func ttUpdate(c *cli.Context) error {
}
if !updated {
fatal("Nothing to update. Use --cron or --function.")
log.Fatal("Nothing to update. Use --cron or --function.")
}
_, err = client.TimeTriggerUpdate(tt)
@@ -161,7 +162,7 @@ func ttDelete(c *cli.Context) error {
client := getClient(c.GlobalString("server"))
ttName := c.String("name")
if len(ttName) == 0 {
fatal("Need name of trigger to delete, use --name")
log.Fatal("Need name of trigger to delete, use --name")
}
ttNs := c.String("triggerns")
@@ -200,7 +201,7 @@ func ttTest(c *cli.Context) error {
round := c.Int("round")
cronSpec := c.String("cron")
if len(cronSpec) == 0 {
fatal("Need a cron spec like '0 30 * * * *', '@every 1h30m', or '@hourly'; use --cron")
log.Fatal("Need a cron spec like '0 30 * * * *', '@every 1h30m', or '@hourly'; use --cron")
}
err := getCronNextNActivationTime(cronSpec, getAPITimeInfo(client), round)
fission/upgrade.go
+4 -3
View File
@@ -16,6 +16,7 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
"github.com/fission/fission/v1"
)
@@ -37,7 +38,7 @@ type (
func getV1URL(serverUrl string) string {
if len(serverUrl) == 0 {
fatal("Need --server or FISSION_URL set to your fission server.")
log.Fatal("Need --server or FISSION_URL set to your fission server.")
}
isHTTPS := strings.Index(serverUrl, "https://") == 0
isHTTP := strings.Index(serverUrl, "http://") == 0
@@ -57,7 +58,7 @@ func get(url string) []byte {
checkErr(err, "reading server response")
if resp.StatusCode != 200 {
fatal(fmt.Sprintf("Failed to fetch fission v0.1 state: %v", string(body)))
log.Fatal(fmt.Sprintf("Failed to fetch fission v0.1 state: %v", string(body)))
}
return body
}
@@ -251,7 +252,7 @@ func upgradeDumpState(c *cli.Context) error {
checkErr(err, "reach fission server")
if resp.StatusCode == http.StatusNotFound {
msg := fmt.Sprintf("Server %v isn't a v1 Fission server. Use --server to point at a pre-0.2.x Fission server.", u)
fatal(msg)
log.Fatal(msg)
}
upgradeDumpV1State(u, filename)
fission/watch.go
+5 -4
View File
@@ -27,6 +27,7 @@ import (
"github.com/fission/fission"
"github.com/fission/fission/crd"
"github.com/fission/fission/fission/log"
)
func wCreate(c *cli.Context) error {
@@ -34,7 +35,7 @@ func wCreate(c *cli.Context) error {
fnName := c.String("function")
if len(fnName) == 0 {
fatal("Need a function name to create a watch, use --function")
log.Fatal("Need a function name to create a watch, use --function")
}
fnNamespace := c.String("fnNamespace")
@@ -95,13 +96,13 @@ func wCreate(c *cli.Context) error {
func wGet(c *cli.Context) error {
// TODO
fatal("Not implemented")
log.Fatal("Not implemented")
return nil
}
func wUpdate(c *cli.Context) error {
// TODO
fatal("Not implemented")
log.Fatal("Not implemented")
return nil
}
@@ -110,7 +111,7 @@ func wDelete(c *cli.Context) error {
wName := c.String("name")
if len(wName) == 0 {
fatal("Need name of watch to delete, use --name")
log.Fatal("Need name of watch to delete, use --name")
}
wNs := c.String("triggerns")