Merge pull request #99 from fuweid/support-pod-log

api: support GetPodLog request type

api/types/load_traffic.go

@@ -80,6 +80,8 @@ type WeightedRequest struct {
 	QuorumGet *RequestGet `json:"quorumGet,omitempty" yaml:"quorumGet,omitempty"`
 	// Put means this is mutating request.
 	Put *RequestPut `json:"put,omitempty" yaml:"put,omitempty"`
+	// GetPodLog means this is to get log from target pod.
+	GetPodLog *RequestGetPodLog `json:"getPodLog,omitempty" yaml:"getPodLog,omitempty"`
 }
 
 // RequestGet defines GET request for target object.
@@ -124,6 +126,23 @@ type RequestPut struct {
 	ValueSize int `json:"valueSize" yaml:"valueSize"`
 }
 
+// RequestGetPodLog defines GetLog request for target pod.
+type RequestGetPodLog struct {
+	// Namespace is pod's namespace.
+	Namespace string `json:"namespace" yaml:"namespace"`
+	// Name is pod's name.
+	Name string `json:"name" yaml:"name"`
+	// Container is target for stream logs. If empty, it's only valid
+	// when there is only one container.
+	Container string `json:"container" yaml:"container"`
+	// TailLines is the number of lines from the end of the logs to show,
+	// if set.
+	TailLines *int64 `json:"tailLines" yaml:"tailLines"`
+	// LimitBytes is the number of bytes to read from the server before
+	// terminating the log output, if set.
+	LimitBytes *int64 `json:"limitBytes" yaml:"limitBytes"`
+}
+
 // Validate verifies fields of LoadProfile.
 func (lp LoadProfile) Validate() error {
 	if lp.Version != 1 {
@@ -180,6 +199,8 @@ func (r WeightedRequest) Validate() error {
 		return r.QuorumGet.Validate()
 	case r.Put != nil:
 		return r.Put.Validate()
+	case r.GetPodLog != nil:
+		return r.GetPodLog.Validate()
 	default:
 		return fmt.Errorf("empty request value")
 	}
@@ -228,6 +249,17 @@ func (r *RequestPut) Validate() error {
 	return nil
 }
 
+// Validate validates RequestGetPodLog type.
+func (r *RequestGetPodLog) Validate() error {
+	if r.Namespace == "" {
+		return fmt.Errorf("namespace is required")
+	}
+	if r.Name == "" {
+		return fmt.Errorf("name is required")
+	}
+	return nil
+}
+
 // Validate validates KubeGroupVersionResource.
 func (m *KubeGroupVersionResource) Validate() error {
 	if m.Version == "" {

api/types/load_traffic_test.go

@@ -56,6 +56,13 @@ spec:
       keySpaceSize: 1000
       valueSize: 1024
     shares: 1000
+  - getPodLog:
+      namespace: default
+      name: hello
+      container: main
+      tailLines: 1000
+      limitBytes: 1024
+    shares: 10
 `
 
 	target := LoadProfile{}
@@ -65,7 +72,7 @@ spec:
 	assert.Equal(t, float64(100), target.Spec.Rate)
 	assert.Equal(t, 10000, target.Spec.Total)
 	assert.Equal(t, 2, target.Spec.Conns)
-	assert.Len(t, target.Spec.Requests, 5)
+	assert.Len(t, target.Spec.Requests, 6)
 
 	assert.Equal(t, 100, target.Spec.Requests[0].Shares)
 	assert.NotNil(t, target.Spec.Requests[0].StaleGet)
@@ -99,6 +106,14 @@ spec:
 	assert.Equal(t, "kperf-", target.Spec.Requests[4].Put.Name)
 	assert.Equal(t, 1000, target.Spec.Requests[4].Put.KeySpaceSize)
 	assert.Equal(t, 1024, target.Spec.Requests[4].Put.ValueSize)
+
+	assert.Equal(t, 10, target.Spec.Requests[5].Shares)
+	assert.NotNil(t, target.Spec.Requests[5].GetPodLog)
+	assert.Equal(t, "default", target.Spec.Requests[5].GetPodLog.Namespace)
+	assert.Equal(t, "hello", target.Spec.Requests[5].GetPodLog.Name)
+	assert.Equal(t, "main", target.Spec.Requests[5].GetPodLog.Container)
+	assert.Equal(t, int64(1000), *target.Spec.Requests[5].GetPodLog.TailLines)
+	assert.Equal(t, int64(1024), *target.Spec.Requests[5].GetPodLog.LimitBytes)
 }
 
 func TestWeightedRequest(t *testing.T) {

request/random.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/Azure/kperf/api/types"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -48,6 +49,8 @@ func NewWeightedRandomRequests(spec *types.LoadProfileSpec) (*WeightedRandomRequ
 			builder = newRequestGetBuilder(r.StaleGet, "0", spec.MaxRetries)
 		case r.QuorumGet != nil:
 			builder = newRequestGetBuilder(r.QuorumGet, "", spec.MaxRetries)
+		case r.GetPodLog != nil:
+			builder = newRequestGetPodLogBuilder(r.GetPodLog, spec.MaxRetries)
 		default:
 			return nil, fmt.Errorf("not implement for PUT yet")
 		}
@@ -221,3 +224,54 @@ func (b *requestListBuilder) Build(cli rest.Interface) (string, *rest.Request) {
 			schema.GroupVersion{Version: "v1"},
 		).MaxRetries(b.maxRetries)
 }
+
+type requestGetPodLogBuilder struct {
+	namespace  string
+	name       string
+	container  string
+	tailLines  *int64
+	limitBytes *int64
+	maxRetries int
+}
+
+func newRequestGetPodLogBuilder(src *types.RequestGetPodLog, maxRetries int) *requestGetPodLogBuilder {
+	b := &requestGetPodLogBuilder{
+		namespace:  src.Namespace,
+		name:       src.Name,
+		container:  src.Container,
+		maxRetries: maxRetries,
+	}
+	if src.TailLines != nil {
+		b.tailLines = toPtr(*src.TailLines)
+	}
+	if src.LimitBytes != nil {
+		b.limitBytes = toPtr(*src.LimitBytes)
+	}
+	return b
+}
+
+// Build implements RequestBuilder.Build.
+func (b *requestGetPodLogBuilder) Build(cli rest.Interface) (string, *rest.Request) {
+	// https://kubernetes.io/docs/reference/using-api/#api-groups
+	apiPath, version := "api", "v1"
+
+	comps := make([]string, 2, 7)
+	comps[0], comps[1] = apiPath, version
+	comps = append(comps, "namespaces", b.namespace)
+	comps = append(comps, "pods", b.name, "log")
+
+	return "POD_LOG", cli.Get().AbsPath(comps...).
+		SpecificallyVersionedParams(
+			&corev1.PodLogOptions{
+				Container:  b.container,
+				TailLines:  b.tailLines,
+				LimitBytes: b.limitBytes,
+			},
+			scheme.ParameterCodec,
+			schema.GroupVersion{Version: "v1"},
+		).MaxRetries(b.maxRetries)
+}
+
+func toPtr[T any](v T) *T {
+	return &v
+}