Use kmeshctl to control accesslog on and off.

Signed-off-by: LiZhenCheng9527 <[email protected]>

ctl/accesslog/accesslog.go

@@ -0,0 +1,105 @@
+/*
+ * Copyright The Kmesh 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 accesslog
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+
+	"github.com/spf13/cobra"
+
+	"kmesh.net/kmesh/ctl/utils"
+	"kmesh.net/kmesh/pkg/logger"
+)
+
+const (
+	patternAccesslog = "/accesslog"
+)
+
+var log = logger.NewLoggerScope("kmeshctl/accesslog")
+
+func NewCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "accesslog",
+		Short:   "enable or disable Kmesh's accesslog",
+		Example: `# enable Kmesh's accesslog: kmeshctl accesslog <kmesh-daemon-pod> enable`,
+		Args:    cobra.MinimumNArgs(1),
+		Run: func(cmd *cobra.Command, args []string) {
+			SetAccesslog(cmd, args)
+		},
+	}
+	return cmd
+}
+
+func SetAccesslog(cmd *cobra.Command, args []string) {
+	podName := args[0]
+	accesslogFlag := args[1]
+	if accesslogFlag != "true" && accesslogFlag != "false" {
+		log.Errorf("Error: Argument must be 'true' or 'false'")
+		os.Exit(1)
+	}
+
+	fw, err := utils.CreateKmeshPortForwarder(podName)
+	if err != nil {
+		log.Errorf("failed to create port forwarder for Kmesh daemon pod %s: %v", podName, err)
+		os.Exit(1)
+	}
+	if err := fw.Start(); err != nil {
+		log.Errorf("failed to start port forwarder for Kmesh daemon pod %s: %v", podName, err)
+		os.Exit(1)
+	}
+	defer fw.Close()
+
+	data, err := json.Marshal(accesslogFlag)
+	if err != nil {
+		log.Errorf("Error marshaling accesslog info: %v", err)
+		return
+	}
+
+	url := fmt.Sprintf("http://%s%s", fw.Address(), patternAccesslog)
+
+	req, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(data))
+	if err != nil {
+		log.Errorf("Error creating request: %v", err)
+		return
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		log.Errorf("failed to make HTTP request: %v", err)
+		return
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		log.Errorf("Error: received status code %d", resp.StatusCode)
+		return
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		log.Errorf("failed to read HTTP response body: %v", err)
+		return
+	}
+	fmt.Println(string(body))
+}

ctl/main.go

@@ -21,6 +21,7 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"kmesh.net/kmesh/ctl/accesslog"
 	"kmesh.net/kmesh/ctl/dump"
 	logcmd "kmesh.net/kmesh/ctl/log"
 	"kmesh.net/kmesh/ctl/waypoint"
@@ -39,6 +40,7 @@ func main() {
 	rootCmd.AddCommand(logcmd.NewCmd())
 	rootCmd.AddCommand(dump.NewCmd())
 	rootCmd.AddCommand(waypoint.NewCmd())
+	rootCmd.AddCommand(accesslog.NewCmd())
 
 	if err := rootCmd.Execute(); err != nil {
 		os.Exit(1)

deploy/yaml/kmesh.yaml

@@ -69,12 +69,12 @@ spec:
                   path: istio-token
       containers:
         - name: kmesh
-          image: ghcr.io/kmesh-net/kmesh:latest
+          image: ghcr.io/kmesh-net/kmesh:7.5.2
           imagePullPolicy: IfNotPresent
           command: ["/bin/sh", "-c"]
           args:
             [
-              "./start_kmesh.sh --mode=workload --enable-bypass=false --enable-bpf-log=true",
+              "./start_kmesh.sh --mode=workload --enable-bypass=false --enable-bpf-log=true --enable-accesslog=true",
             ]
           securityContext:
             privileged: true

pkg/controller/telemetry/metric.go

@@ -50,7 +50,7 @@ const (
 var osStartTime time.Time
 
 type MetricController struct {
-	enableAccesslog     bool
+	EnableAccesslog     bool
 	workloadCache       cache.WorkloadCache
 	workloadMetricCache map[workloadMetricLabels]*workloadMetricInfo
 	serviceMetricCache  map[serviceMetricLabels]*serviceMetricInfo
@@ -176,7 +176,7 @@ type serviceMetricLabels struct {
 
 func NewMetric(workloadCache cache.WorkloadCache, enableAccesslog bool) *MetricController {
 	return &MetricController{
-		enableAccesslog:     enableAccesslog,
+		EnableAccesslog:     enableAccesslog,
 		workloadCache:       workloadCache,
 		workloadMetricCache: map[workloadMetricLabels]*workloadMetricInfo{},
 		serviceMetricCache:  map[serviceMetricLabels]*serviceMetricInfo{},
@@ -262,7 +262,7 @@ func (m *MetricController) Run(ctx context.Context, mapOfTcpInfo *ebpf.Map) {
 				serviceLabels.reporter = "source"
 				accesslog.direction = "OUTBOUND"
 			}
-			if data.state == TCP_CLOSTED && accesslog.sourceWorkload != "-" && m.enableAccesslog {
+			if data.state == TCP_CLOSTED && accesslog.sourceWorkload != "-" && m.EnableAccesslog {
 				OutputAccesslog(data, accesslog)
 			}
 			m.mutex.Lock()

pkg/status/status_server.go

@@ -55,6 +55,7 @@ const (
 	patternConfigDumpWorkload = configDumpPrefix + "/workload"
 	patternReadyProbe         = "/debug/ready"
 	patternLoggers            = "/debug/loggers"
+	patternAccesslog          = "/accesslog"
 
 	bpfLoggerName = "bpf"
 
@@ -100,6 +101,7 @@ func NewServer(c *controller.XdsClient, configs *options.BootstrapConfigs, bpfLo
 	s.mux.HandleFunc(patternConfigDumpAds, s.configDumpAds)
 	s.mux.HandleFunc(patternConfigDumpWorkload, s.configDumpWorkload)
 	s.mux.HandleFunc(patternLoggers, s.loggersHandler)
+	s.mux.HandleFunc(patternAccesslog, s.accesslogHandler)
 
 	// TODO: add dump certificate, authorizationPolicies and services
 	s.mux.HandleFunc(patternReadyProbe, s.readyProbe)
@@ -236,6 +238,30 @@ func (s *Server) loggersHandler(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (s *Server) accesslogHandler(w http.ResponseWriter, r *http.Request) {
+	var accesslogInfo string
+
+	defer r.Body.Close()
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		fmt.Fprintf(w, "\t%s: %v\n", "Error reading request body", err)
+		return
+	}
+
+	if err := json.Unmarshal(body, &accesslogInfo); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		fmt.Fprintf(w, "\t%s: %v\n", "Invalid request body format", err)
+		return
+	}
+
+	if accesslogInfo == "true" {
+		s.xdsClient.WorkloadController.MetricController.EnableAccesslog = true
+	} else {
+		s.xdsClient.WorkloadController.MetricController.EnableAccesslog = false
+	}
+}
+
 func (s *Server) getLoggerNames(w http.ResponseWriter) {
 	loggerNames := append(logger.GetLoggerNames(), bpfLoggerName)
 	data, err := json.MarshalIndent(&loggerNames, "", "    ")