Merge pull request #304 from meshery/MUzairS15/fix/log-session

fix init and close of logs stream

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/layer5io/meshery-operator v0.7.0
 	github.com/layer5io/meshkit v0.7.10
 	github.com/myntra/pipeline v0.0.0-20180618182531-2babf4864ce8
+	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/viper v1.18.2
 	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 	golang.org/x/net v0.20.0
@@ -140,7 +141,6 @@ require (
 	github.com/sagikazarmark/locafero v0.4.0 // indirect
 	github.com/sagikazarmark/slog-shim v0.1.0 // indirect
 	github.com/shopspring/decimal v1.3.1 // indirect
-	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.11.0 // indirect
 	github.com/spf13/cast v1.6.0 // indirect

internal/config/crd_config.go

@@ -3,6 +3,7 @@ package config
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/layer5io/meshery-operator/pkg/client"
 	"github.com/layer5io/meshkit/utils"
@@ -168,7 +169,7 @@ func PopulateConfigs(configMap corev1.ConfigMap) (*MeshsyncConfig, error) {
 func PatchCRVersion(config *rest.Config) error {
 	meshsyncClient, err := client.New(config)
 	if err != nil {
-		return err
+		return ErrInitConfig(fmt.Errorf("unable to update MeshSync configuration"))
 	}
 
 	patchedResource := map[string]interface{}{
@@ -178,11 +179,11 @@ func PatchCRVersion(config *rest.Config) error {
 	}
 	byt, err := utils.Marshal(patchedResource)
 	if err != nil {
-		return err
+		return ErrInitConfig(fmt.Errorf("unable to update MeshSync configuration"))
 	}
 	_, err = meshsyncClient.CoreV1Alpha1().MeshSyncs("meshery").Patch(context.TODO(), crName, types.MergePatchType, []byte(byt), metav1.PatchOptions{})
 	if err != nil {
-		return err
+		return ErrInitConfig(fmt.Errorf("unable to update MeshSync configuration"))
 	}
 	return nil
 }

main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/layer5io/meshsync/internal/channels"
 	"github.com/layer5io/meshsync/internal/config"
 	"github.com/layer5io/meshsync/meshsync"
+	"github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
 )
 
@@ -33,7 +34,8 @@ func main() {
 
 	// Initialize Logger instance
 	log, err := logger.New(serviceName, logger.Options{
-		Format: logger.SyslogLogFormat,
+		Format:   logger.SyslogLogFormat,
+		LogLevel: int(logrus.InfoLevel),
 	})
 	if err != nil {
 		fmt.Println(err)

meshsync/logstream.go

@@ -62,17 +62,22 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene
 		return
 	}
 
-	defer resp.Close()
+	go func() {
+		<-h.channelPool[id].(channels.StructChannel)
+		h.Log.Info("Closing", id)
+		delete(h.channelPool, id)
+		resp.Close()
+	}()
 
 	for {
 		buf := make([]byte, 2000)
 		numBytes, err := resp.Read(buf)
-		if numBytes == 0 {
-			continue
-		}
 		if err == io.EOF {
 			break
 		}
+		if numBytes == 0 {
+			continue
+		}
 		if err != nil {
 			h.Log.Error(ErrCopyBuffer(err))
 			delete(h.channelPool, id)
@@ -94,7 +99,4 @@ func (h *Handler) streamLogs(id string, req model.LogRequest, cfg config.Listene
 		}
 	}
 
-	<-h.channelPool[id].(channels.StructChannel)
-	h.Log.Info("Closing", id)
-	delete(h.channelPool, id)
 }