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

Make CLI functions return error instead of fatal out (#1379)

Browse Source 
Before this PR, CLI functions fatal out when encountering error
instead of returning it. Such behavior makes it hard to reuse 
the functions nor writing unit tests. This PR aims to make functions 
return errors instead of error out.
This commit is contained in:
Ta-Ching Chen
2019-11-05 16:50:13 +08:00
committed by GitHub
parent b0d27ee5d2
commit 93d4b88d84
72 changed files with 894 additions and 632 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/fission-cli/util/portforward.go
+31 -26
View File
@@ -25,12 +25,13 @@ import (
"strings"
"time"
"github.com/pkg/errors"
v1 "k8s.io/api/core/v1"
meta_v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/tools/portforward"
"k8s.io/client-go/transport/spdy"
"github.com/fission/fission/pkg/fission-cli/log"
"github.com/fission/fission/pkg/fission-cli/consolemsg"
"github.com/fission/fission/pkg/utils"
)
@@ -39,16 +40,16 @@ import (
// 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(namespace, labelSelector string) string {
log.Verbose(2, "Setting up port forward to %s in namespace %s",
func SetupPortForward(namespace, labelSelector string) (string, error) {
consolemsg.Verbose(2, "Setting up port forward to %s in namespace %s",
labelSelector, namespace)
localPort, err := findFreePort()
if err != nil {
log.Fatal(fmt.Sprintf("Error finding unused port :%v", err.Error()))
return "", errors.Wrap(err, "error finding unused port")
}
log.Verbose(2, "Waiting for local port %v", localPort)
consolemsg.Verbose(2, "Waiting for local port %v", localPort)
for {
conn, _ := net.DialTimeout("tcp",
net.JoinHostPort("", localPort), time.Millisecond)
@@ -60,15 +61,16 @@ func SetupPortForward(namespace, labelSelector string) string {
time.Sleep(time.Millisecond * 50)
}
log.Verbose(2, "Starting port forward from local port %v", localPort)
consolemsg.Verbose(2, "Starting port forward from local port %v", localPort)
go func() {
err := runPortForward(labelSelector, localPort, namespace)
if err != nil {
log.Fatal(fmt.Sprintf("Error forwarding to port %v: %s", localPort, err.Error()))
fmt.Printf("Error forwarding to port %v: %s", localPort, err.Error())
os.Exit(1)
}
}()
log.Verbose(2, "Waiting for port forward %v to start...", localPort)
consolemsg.Verbose(2, "Waiting for port forward %v to start...", localPort)
for {
conn, _ := net.DialTimeout("tcp",
net.JoinHostPort("", localPort), time.Millisecond)
@@ -79,9 +81,9 @@ func SetupPortForward(namespace, labelSelector string) string {
time.Sleep(time.Millisecond * 50)
}
log.Verbose(2, "Port forward from local port %v started", localPort)
consolemsg.Verbose(2, "Port forward from local port %v started", localPort)
return localPort
return localPort, nil
}
func findFreePort() (string, error) {
@@ -102,9 +104,12 @@ func findFreePort() (string, error) {
// runPortForward creates a local port forward to the specified pod
func runPortForward(labelSelector string, localPort string, ns string) error {
config, clientset := GetKubernetesClient()
config, clientset, err := GetKubernetesClient()
if err != nil {
return err
}
log.Verbose(2, "Connected to Kubernetes API")
consolemsg.Verbose(2, "Connected to Kubernetes API")
// if namespace is unset, try to find a pod in any namespace
if len(ns) == 0 {
@@ -114,8 +119,10 @@ func runPortForward(labelSelector string, localPort string, ns string) error {
// get the pod; if there is more than one, ask the user to disambiguate
podList, err := clientset.CoreV1().Pods(ns).
List(meta_v1.ListOptions{LabelSelector: labelSelector})
if err != nil || len(podList.Items) == 0 {
log.Fatal(fmt.Sprintf("Error getting pod for port-forwarding with label selector %v: %v", labelSelector, err))
if err != nil {
return errors.Wrapf(err, "error getting pod for port-forwarding with label selector %v", labelSelector)
} else if len(podList.Items) == 0 {
return errors.Errorf("no available pod for port-forwarding with label selector %v", labelSelector)
}
nsList := make([]string, 0)
@@ -131,8 +138,8 @@ func runPortForward(labelSelector string, localPort string, ns string) error {
namespaces[p.Namespace] = append(namespaces[p.Namespace], &p)
}
if len(nsList) > 1 {
log.Fatal(fmt.Sprintf("Found %v fission installs, set FISSION_NAMESPACE to one of: %v",
len(namespaces), strings.Join(nsList, " ")))
return errors.Errorf("Found %v fission installs, set FISSION_NAMESPACE to one of: %v",
len(namespaces), strings.Join(nsList, " "))
}
}
@@ -141,7 +148,7 @@ func runPortForward(labelSelector string, localPort string, ns string) error {
ns = nsList[0]
pods, ok := namespaces[ns]
if !ok {
log.Fatal(fmt.Sprintf("Error finding fission install within the given namespace %v, please check FISSION_NAMESPACE is set properly", ns))
return errors.Errorf("Error finding fission install within the given namespace %v, please check FISSION_NAMESPACE is set properly", ns)
}
var podName, podNameSpace string
@@ -159,10 +166,10 @@ func runPortForward(labelSelector string, localPort string, ns string) error {
svcs, err := clientset.CoreV1().Services(podNameSpace).
List(meta_v1.ListOptions{LabelSelector: labelSelector})
if err != nil {
log.Fatal(fmt.Sprintf("Error getting %v service :%v", labelSelector, err.Error()))
return errors.Wrapf(err, "Error getting %v service", labelSelector)
}
if len(svcs.Items) == 0 {
log.Fatal(fmt.Sprintf("Service %v not found", labelSelector))
return errors.Errorf("Service %v not found", labelSelector)
}
service := &svcs.Items[0]
@@ -170,7 +177,7 @@ func runPortForward(labelSelector string, localPort string, ns string) error {
for _, servicePort := range service.Spec.Ports {
targetPort = servicePort.TargetPort.String()
}
log.Verbose(2, "Connecting to port %v on pod %v/%v", targetPort, podNameSpace, podNameSpace)
consolemsg.Verbose(2, "Connecting to port %v on pod %v/%v", targetPort, podNameSpace, podNameSpace)
stopChannel := make(chan struct{}, 1)
readyChannel := make(chan struct{})
@@ -187,21 +194,19 @@ func runPortForward(labelSelector string, localPort string, ns string) error {
// actually start the port-forwarding process here
transport, upgrader, err := spdy.RoundTripperFor(config)
if err != nil {
msg := fmt.Sprintf("Failed to connect to Fission service on Kubernetes: %v", err.Error())
log.Fatal(msg)
return errors.Errorf("Failed to connect to Fission service on Kubernetes")
}
dialer := spdy.NewDialer(upgrader, &http.Client{Transport: transport}, "POST", url)
outStream := os.Stdout
if log.Verbosity < 2 {
if consolemsg.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())
log.Fatal(msg)
return errors.Wrap(err, "error creating port forwarder")
}
log.Verbose(2, "Starting port forwarder")
consolemsg.Verbose(2, "Starting port forwarder")
return fw.ForwardPorts()
}
pkg/fission-cli/util/util.go
+153 -45
View File
@@ -22,62 +22,46 @@ import (
"os/user"
"path/filepath"
"regexp"
"strconv"
"strings"
"github.com/hashicorp/go-multierror"
"github.com/pkg/errors"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
restclient "k8s.io/client-go/rest"
"k8s.io/client-go/tools/clientcmd"
"github.com/fission/fission/pkg/controller/client"
"github.com/fission/fission/pkg/fission-cli/log"
"github.com/fission/fission/pkg/fission-cli/cliwrapper/cli"
"github.com/fission/fission/pkg/fission-cli/consolemsg"
"github.com/fission/fission/pkg/fission-cli/flag"
"github.com/fission/fission/pkg/info"
"github.com/fission/fission/pkg/plugin"
)
func GetApiClient(serverUrl string) *client.Client {
if len(serverUrl) == 0 {
// starts local portforwarder etc.
serverUrl = GetServerUrl()
}
isHTTPS := strings.Index(serverUrl, "https://") == 0
isHTTP := strings.Index(serverUrl, "http://") == 0
if !(isHTTP || isHTTPS) {
serverUrl = "http://" + serverUrl
}
return client.MakeClient(serverUrl)
}
func GetFissionNamespace() string {
fissionNamespace := os.Getenv("FISSION_NAMESPACE")
return fissionNamespace
}
func GetServerUrl() string {
return GetApplicationUrl("application=fission-api")
}
func GetApplicationUrl(selector string) string {
func GetApplicationUrl(selector string) (string, error) {
var serverUrl string
// Use FISSION_URL env variable if set; otherwise, port-forward to controller.
fissionUrl := os.Getenv("FISSION_URL")
if len(fissionUrl) == 0 {
fissionNamespace := GetFissionNamespace()
localPort := SetupPortForward(fissionNamespace, "application=fission-api")
localPort, err := SetupPortForward(fissionNamespace, selector)
if err != nil {
return "", err
}
serverUrl = "http://127.0.0.1:" + localPort
} else {
serverUrl = fissionUrl
}
return serverUrl
}
func CheckErr(err error, msg string) {
if err != nil {
log.Fatal(fmt.Sprintf("Failed to %v: %v", msg, err))
}
return serverUrl, nil
}
// KubifyName make a kubernetes compliant name out of an arbitrary string
@@ -88,18 +72,15 @@ func KubifyName(old string) string {
newName := strings.ToLower(old)
// replace disallowed chars with '-'
inv, err := regexp.Compile("[^-a-z0-9]")
CheckErr(err, "compile regexp")
inv, _ := regexp.Compile("[^-a-z0-9]")
newName = string(inv.ReplaceAll([]byte(newName), []byte("-")))
// trim leading non-alphabetic
leadingnonalpha, err := regexp.Compile("^[^a-z]+")
CheckErr(err, "compile regexp")
leadingnonalpha, _ := regexp.Compile("^[^a-z]+")
newName = string(leadingnonalpha.ReplaceAll([]byte(newName), []byte{}))
// trim trailing
trailing, err := regexp.Compile("[^a-z0-9]+$")
CheckErr(err, "compile regexp")
trailing, _ := regexp.Compile("[^a-z0-9]+$")
newName = string(trailing.ReplaceAll([]byte(newName), []byte{}))
// truncate to length
@@ -119,7 +100,7 @@ func KubifyName(old string) string {
// GetKubernetesClient builds a new kubernetes client. If the KUBECONFIG
// environment variable is empty or doesn't exist, ~/.kube/config is used for
// the kube config path
func GetKubernetesClient() (*restclient.Config, *kubernetes.Clientset) {
func GetKubernetesClient() (*restclient.Config, *kubernetes.Clientset, error) {
loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
kubeConfigPath := os.Getenv("KUBECONFIG")
@@ -130,7 +111,7 @@ func GetKubernetesClient() (*restclient.Config, *kubernetes.Clientset) {
// In case that user.Current() may be unable to work under some circumstances and return errors like
// "user: Current not implemented on darwin/amd64" due to cross-compilation problem. (https://github.com/golang/go/issues/6376).
// Instead of doing fatal here, we fallback to get home directory from the environment $HOME.
log.Warn(fmt.Sprintf("Could not get the current user's directory (%s), fallback to get it from env $HOME", err))
consolemsg.Warn(fmt.Sprintf("Could not get the current user's directory (%s), fallback to get it from env $HOME", err))
homeDir = os.Getenv("HOME")
} else {
homeDir = usr.HomeDir
@@ -138,27 +119,27 @@ func GetKubernetesClient() (*restclient.Config, *kubernetes.Clientset) {
kubeConfigPath = filepath.Join(homeDir, ".kube", "config")
if _, err := os.Stat(kubeConfigPath); os.IsNotExist(err) {
log.Fatal("Couldn't find kubeconfig file. " +
return nil, nil, errors.New("Couldn't find kubeconfig file. " +
"Set the KUBECONFIG environment variable to your kubeconfig's path.")
}
loadingRules.ExplicitPath = kubeConfigPath
log.Verbose(2, "Using kubeconfig from %q", kubeConfigPath)
consolemsg.Verbose(2, "Using kubeconfig from %q", kubeConfigPath)
} else {
log.Verbose(2, "Using kubeconfig from environment %q", kubeConfigPath)
consolemsg.Verbose(2, "Using kubeconfig from environment %q", kubeConfigPath)
}
config, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
loadingRules, &clientcmd.ConfigOverrides{}).ClientConfig()
if err != nil {
log.Fatal(fmt.Sprintf("Failed to build Kubernetes config: %s", err))
return nil, nil, errors.Wrap(err, "Failed to build Kubernetes config")
}
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
log.Fatal(fmt.Sprintf("Failed to connect to Kubernetes: %s", err))
return nil, nil, errors.Wrap(err, "Failed to connect to Kubernetes")
}
return config, clientset
return config, clientset, nil
}
// given a list of functions, this checks if the functions actually exist on the cluster
@@ -199,7 +180,7 @@ func GetVersion(client *client.Client) info.Versions {
serverInfo, err := client.ServerInfo()
if err != nil {
log.Warn(fmt.Sprintf("Error getting Fission API version: %v", err))
consolemsg.Warn(fmt.Sprintf("Error getting Fission API version: %v", err))
serverInfo = &info.ServerInfo{}
}
@@ -212,3 +193,130 @@ func GetVersion(client *client.Client) info.Versions {
return versions
}
func GetServer(flags cli.Input) (c *client.Client, err error) {
serverUrl := flags.GlobalString(flag.FISSION_SERVER)
if len(serverUrl) == 0 {
// starts local portforwarder etc.
serverUrl, err = GetApplicationUrl("application=fission-api")
if err != nil {
return nil, err
}
}
isHTTPS := strings.Index(serverUrl, "https://") == 0
isHTTP := strings.Index(serverUrl, "http://") == 0
if !(isHTTP || isHTTPS) {
serverUrl = "http://" + serverUrl
}
return client.MakeClient(serverUrl), nil
}
func GetResourceReqs(flags cli.Input, resReqs *v1.ResourceRequirements) (*v1.ResourceRequirements, error) {
r := &v1.ResourceRequirements{}
if resReqs != nil {
r.Requests = resReqs.Requests
r.Limits = resReqs.Limits
}
if len(r.Requests) == 0 {
r.Requests = make(map[v1.ResourceName]resource.Quantity)
}
if len(r.Limits) == 0 {
r.Limits = make(map[v1.ResourceName]resource.Quantity)
}
e := &multierror.Error{}
if flags.IsSet(flag.RUNTIME_MINCPU) {
mincpu := flags.Int(flag.RUNTIME_MINCPU)
cpuRequest, err := resource.ParseQuantity(strconv.Itoa(mincpu) + "m")
if err != nil {
e = multierror.Append(e, errors.Wrap(err, "Failed to parse mincpu"))
}
r.Requests[v1.ResourceCPU] = cpuRequest
}
if flags.IsSet(flag.RUNTIME_MINMEMORY) {
minmem := flags.Int(flag.RUNTIME_MINMEMORY)
memRequest, err := resource.ParseQuantity(strconv.Itoa(minmem) + "Mi")
if err != nil {
e = multierror.Append(e, errors.Wrap(err, "Failed to parse minmemory"))
}
r.Requests[v1.ResourceMemory] = memRequest
}
if flags.IsSet(flag.RUNTIME_MAXCPU) {
maxcpu := flags.Int(flag.RUNTIME_MAXCPU)
cpuLimit, err := resource.ParseQuantity(strconv.Itoa(maxcpu) + "m")
if err != nil {
e = multierror.Append(e, errors.Wrap(err, "Failed to parse maxcpu"))
}
r.Limits[v1.ResourceCPU] = cpuLimit
}
if flags.IsSet(flag.RUNTIME_MAXMEMORY) {
maxmem := flags.Int(flag.RUNTIME_MAXMEMORY)
memLimit, err := resource.ParseQuantity(strconv.Itoa(maxmem) + "Mi")
if err != nil {
e = multierror.Append(e, errors.Wrap(err, "Failed to parse maxmemory"))
}
r.Limits[v1.ResourceMemory] = memLimit
}
limitCPU := r.Limits[v1.ResourceCPU]
requestCPU := r.Requests[v1.ResourceCPU]
if limitCPU.IsZero() && !requestCPU.IsZero() {
r.Limits[v1.ResourceCPU] = requestCPU
} else if limitCPU.Cmp(requestCPU) < 0 {
e = multierror.Append(e, fmt.Errorf("MinCPU (%v) cannot be greater than MaxCPU (%v)", requestCPU.String(), limitCPU.String()))
}
limitMem := r.Limits[v1.ResourceMemory]
requestMem := r.Requests[v1.ResourceMemory]
if limitMem.IsZero() && !requestMem.IsZero() {
r.Limits[v1.ResourceMemory] = requestMem
} else if limitMem.Cmp(requestMem) < 0 {
e = multierror.Append(e, fmt.Errorf("MinMemory (%v) cannot be greater than MaxMemory (%v)", requestMem.String(), limitMem.String()))
}
if e.ErrorOrNil() != nil {
return nil, e
}
return &v1.ResourceRequirements{
Requests: r.Requests,
Limits: r.Limits,
}, nil
}
func GetSpecDir(flags cli.Input) string {
specDir := flags.String(flag.SPEC_SPECDIR)
if len(specDir) == 0 {
specDir = "specs"
}
return specDir
}
// GetMetadata returns a pointer to ObjectMeta that is populated with resource name and namespace given by the user.
func GetMetadata(nameFlagText string, namespaceFlagText string, flags cli.Input) (*metav1.ObjectMeta, error) {
name := flags.String(nameFlagText)
if len(name) == 0 {
return nil, errors.Errorf("need a resource name, use --%v", nameFlagText)
}
ns := flags.String(namespaceFlagText)
m := &metav1.ObjectMeta{
Name: name,
Namespace: ns,
}
return m, nil
}