Add --kubernetes flag to make dapr invoke cmd support kubernetes

README.md

@@ -492,6 +492,14 @@ By default, Dapr will use the `POST` verb. If your app uses Dapr for gRPC, you s
 dapr invoke --app-id nodeapp --method mymethod --verb GET
 ```
 
+Invoke your app in Kubernetes mode:
+
+If your app running in a Kubernetes cluster, use the `invoke` command with `--kubernetes` flag or the `-k` shorthand.
+
+```
+$ dapr invoke --kubernetes --app-id nodeapp --method mymethod
+```
+
 ### List
 
 To list all Dapr instances running on your machine:

cmd/invoke.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/dapr/cli/pkg/kubernetes"
 	"github.com/dapr/cli/pkg/print"
 	"github.com/dapr/cli/pkg/standalone"
 )
@@ -38,15 +39,18 @@ var (
 
 var InvokeCmd = &cobra.Command{
 	Use:   "invoke",
-	Short: "Invoke a method on a given Dapr application. Supported platforms: Self-hosted",
+	Short: "Invoke a method on a given Dapr application. Supported platforms: Kubernetes and self-hosted",
 	Example: `
-# Invoke a sample method on target app with POST Verb
-dapr invoke --app-id target --method sample --data '{"key":"value"}
+# Invoke a sample method on target app with POST Verb in self-hosted mode
+dapr invoke --app-id target --method sample --data '{"key":"value"}'
 
-# Invoke a sample method on target app with GET Verb
+# Invoke a sample method on target app with in Kubernetes
+dapr invoke -k --app-id target --method sample --data '{"key":"value"}'
+
+# Invoke a sample method on target app with GET Verb in self-hosted mode
 dapr invoke --app-id target --method sample --verb GET
 
-# Invoke a sample method on target app with GET Verb using Unix domain socket
+# Invoke a sample method on target app with GET Verb using Unix domain socket in self-hosted mode
 dapr invoke --unix-domain-socket --app-id target --method sample --verb GET
 `,
 	Run: func(cmd *cobra.Command, args []string) {
@@ -66,7 +70,6 @@ dapr invoke --unix-domain-socket --app-id target --method sample --verb GET
 		} else if invokeData != "" {
 			bytePayload = []byte(invokeData)
 		}
-		client := standalone.NewClient()
 
 		// TODO(@daixiang0): add Windows support.
 		if invokeSocket != "" {
@@ -78,7 +81,14 @@ dapr invoke --unix-domain-socket --app-id target --method sample --verb GET
 			}
 		}
 
-		response, err := client.Invoke(invokeAppID, invokeAppMethod, bytePayload, invokeVerb, invokeSocket)
+		var response string
+		if kubernetesMode {
+			response, err = kubernetes.Invoke(invokeAppID, invokeAppMethod, bytePayload, invokeVerb)
+		} else {
+			client := standalone.NewClient()
+			response, err = client.Invoke(invokeAppID, invokeAppMethod, bytePayload, invokeVerb, invokeSocket)
+		}
+
 		if err != nil {
 			err = fmt.Errorf("error invoking app %s: %w", invokeAppID, err)
 			print.FailureStatusEvent(os.Stderr, err.Error())
@@ -93,6 +103,7 @@ dapr invoke --unix-domain-socket --app-id target --method sample --verb GET
 }
 
 func init() {
+	InvokeCmd.Flags().BoolVarP(&kubernetesMode, "kubernetes", "k", false, "Invoke a method on a Dapr application in a Kubernetes cluster")
 	InvokeCmd.Flags().StringVarP(&invokeAppID, "app-id", "a", "", "The application id to invoke")
 	InvokeCmd.Flags().StringVarP(&invokeAppMethod, "method", "m", "", "The method to invoke")
 	InvokeCmd.Flags().StringVarP(&invokeData, "data", "d", "", "The JSON serialized data string (optional)")

pkg/kubernetes/invoke.go

@@ -0,0 +1,175 @@
+/*
+Copyright 2021 The Dapr 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 kubernetes
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"strings"
+
+	core_v1 "k8s.io/api/core/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/net"
+	k8s "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+type AppInfo struct {
+	AppID     string `csv:"APP ID"      json:"appId"        yaml:"appId"`
+	HTTPPort  string `csv:"HTTP PORT"   json:"httpPort"     yaml:"httpPort"`
+	GRPCPort  string `csv:"GRPC PORT"   json:"grpcPort"     yaml:"grpcPort"`
+	AppPort   string `csv:"APP PORT"    json:"appPort"      yaml:"appPort"`
+	PodName   string `csv:"POD NAME"    json:"podName"      yaml:"podName"`
+	Namespace string `csv:"NAMESPACE"   json:"namespace"    yaml:"namespace"`
+}
+
+type (
+	DaprPod     core_v1.Pod
+	DaprAppList []*AppInfo
+)
+
+// Invoke is a command to invoke a remote or local dapr instance.
+func Invoke(appID, method string, data []byte, verb string) (string, error) {
+	client, err := Client()
+	if err != nil {
+		return "", err
+	}
+
+	app, err := GetAppInfo(client, appID)
+	if err != nil {
+		return "", err
+	}
+
+	return invoke(client.CoreV1().RESTClient(), app, method, data, verb)
+}
+
+func invoke(client rest.Interface, app *AppInfo, method string, data []byte, verb string) (string, error) {
+	res, err := app.Request(client.Verb(verb), method, data, verb)
+	if err != nil {
+		return "", fmt.Errorf("error get request: %w", err)
+	}
+
+	result := res.Do(context.TODO())
+	rawbody, err := result.Raw()
+	if err != nil {
+		return "", fmt.Errorf("error get raw: %w", err)
+	}
+
+	if len(rawbody) > 0 {
+		return string(rawbody), nil
+	}
+
+	return "", nil
+}
+
+func GetAppInfo(client k8s.Interface, appID string) (*AppInfo, error) {
+	list, err := ListAppInfos(client, appID)
+	if err != nil {
+		return nil, err
+	}
+	if len(list) == 0 {
+		return nil, fmt.Errorf("%s not found", appID)
+	}
+	app := list[0]
+	return app, nil
+}
+
+// List outputs plugins.
+func ListAppInfos(client k8s.Interface, appIDs ...string) (DaprAppList, error) {
+	opts := v1.ListOptions{}
+	podList, err := client.CoreV1().Pods(v1.NamespaceAll).List(context.TODO(), opts)
+	if err != nil {
+		return nil, fmt.Errorf("err get pods list:%w", err)
+	}
+
+	fn := func(*AppInfo) bool {
+		return true
+	}
+	if len(appIDs) > 0 {
+		fn = func(a *AppInfo) bool {
+			for _, id := range appIDs {
+				if id != "" && a.AppID == id {
+					return true
+				}
+			}
+			return false
+		}
+	}
+
+	l := make(DaprAppList, 0)
+	for _, p := range podList.Items {
+		p := DaprPod(p)
+		for _, c := range p.Spec.Containers {
+			if c.Name == "daprd" {
+				app := getAppInfoFromPod(&p)
+				if fn(app) {
+					l = append(l, app)
+				}
+			}
+		}
+	}
+
+	return l, nil
+}
+
+func getAppInfoFromPod(p *DaprPod) *AppInfo {
+	var appInfo *AppInfo
+	for _, c := range p.Spec.Containers {
+		if c.Name == "daprd" {
+			appInfo = &AppInfo{
+				PodName:   p.Name,
+				Namespace: p.Namespace,
+			}
+			for i, arg := range c.Args {
+				switch arg {
+				case "--app-port":
+					appInfo.AppPort = c.Args[i+1]
+				case "--dapr-http-port":
+					appInfo.HTTPPort = c.Args[i+1]
+				case "--dapr-grpc-port":
+					appInfo.GRPCPort = c.Args[i+1]
+				case "--app-id":
+					appInfo.AppID = c.Args[i+1]
+				}
+			}
+		}
+	}
+
+	return appInfo
+}
+
+func (a *AppInfo) Request(r *rest.Request, method string, data []byte, verb string) (*rest.Request, error) {
+	r = r.Namespace(a.Namespace).
+		Resource("pods").
+		SubResource("proxy").
+		SetHeader("Content-Type", "application/json").
+		Name(net.JoinSchemeNamePort("", a.PodName, a.AppPort))
+	if data != nil {
+		r = r.Body(data)
+	}
+
+	u, err := url.Parse(method)
+	if err != nil {
+		return nil, fmt.Errorf("error parse method %s: %w", method, err)
+	}
+
+	r = r.Suffix(u.Path)
+
+	for k, vs := range u.Query() {
+		r = r.Param(k, strings.Join(vs, ","))
+	}
+
+	return r, nil
+}