feat: Server side logs

Makefile

@@ -41,4 +41,8 @@ binaries: linux darwin windows compress
 .PHONY: release
 release: binaries
 	mkdir -p .release
-	cp .bin/$(NAME)* .release/
+	cp .bin/$(NAME)* .release/
+
+.PHONY: integration
+integration:
+	go test --tags=integration ./... -count=1 -v

api/logs/logs.go

@@ -19,6 +19,7 @@ type SearchBackend struct {
 	Routes     []SearchRoute            `json:"routes,omitempty"`
 	Backend    SearchAPI                `json:"-"`
 	Kubernetes *KubernetesSearchBackend `json:"kubernetes,omitempty"`
+	Files      []FileSearchBackend      `json:"file,omitempty" yaml:"file,omitempty"`
 }
 
 type SearchRoute struct {
@@ -34,6 +35,11 @@ type KubernetesSearchBackend struct {
 	Namespace string `json:"namespace,omitempty"`
 }
 
+type FileSearchBackend struct {
+	Labels map[string]string `yaml:"labels,omitempty"`
+	Paths  []string          `yaml:"path,omitempty"`
+}
+
 type SearchParams struct {
 	// Limit is the maximum number of results to return.
 	Limit      int64 `json:"limit,omitempty"`

cmd/serve.go

@@ -37,24 +37,33 @@ func runServe(cmd *cobra.Command, configFiles []string) {
 		}
 	}
 	fmt.Println(logs.GlobalBackends)
+
+	server := SetupServer(kommonsClient)
+	addr := "0.0.0.0:" + strconv.Itoa(httpPort)
+	server.Logger.Fatal(server.Start(addr))
+}
+
+func SetupServer(kClient *kommons.Client) *echo.Echo {
 	e := echo.New()
 	// Extending the context and fetching the kubeconfig client here.
 	// For more info see: https://echo.labstack.com/guide/context/#extending-context
 	e.Use(func(next echo.HandlerFunc) echo.HandlerFunc {
 		return func(c echo.Context) error {
 			cc := &api.Context{
-				Kommons: kommonsClient,
+				Kommons: kClient,
 				Context: c,
 			}
 			return next(cc)
 		}
 	})
+
 	e.GET("/", func(c echo.Context) error {
 		return c.String(http.StatusOK, "apm-hub server running")
 	})
+
 	e.POST("/search", pkg.Search)
-	addr := "0.0.0.0:" + strconv.Itoa(httpPort)
-	e.Logger.Fatal(e.Start(addr))
+
+	return e
 }
 
 func init() {

pkg/config.go

@@ -1,26 +1,29 @@
 package pkg
 
 import (
-	"io/ioutil"
+	"os"
+	"path/filepath"
 
 	"github.com/flanksource/flanksource-ui/apm-hub/api/logs"
+	"github.com/flanksource/flanksource-ui/apm-hub/pkg/files"
 	k8s "github.com/flanksource/flanksource-ui/apm-hub/pkg/kubernetes"
 	"github.com/flanksource/kommons"
 	"gopkg.in/yaml.v3"
 )
 
-//Initilize all the backends mentioned in the config
+// Initilize all the backends mentioned in the config
 // GlobalBackends, error
 func ParseConfig(kommonsClient *kommons.Client, configFile string) ([]logs.SearchBackend, error) {
 	searchConfig := &logs.SearchConfig{}
 	backends := []logs.SearchBackend{}
-	data, err := ioutil.ReadFile(configFile)
+	data, err := os.ReadFile(configFile)
 	if err != nil {
 		return nil, err
 	}
 	if err := yaml.Unmarshal(data, searchConfig); err != nil {
 		return nil, err
 	}
+
 	for _, backend := range searchConfig.Backends {
 		if backend.Kubernetes != nil {
 			client, err := k8s.GetKubeClient(kommonsClient, backend.Kubernetes)
@@ -30,8 +33,26 @@ func ParseConfig(kommonsClient *kommons.Client, configFile string) ([]logs.Searc
 			backend.Backend = &k8s.KubernetesSearch{
 				Client: client,
 			}
+			backends = append(backends, backend)
+		}
+
+		if len(backend.Files) != 0 {
+			// If the paths are not absolute,
+			// They should be parsed with respect to the provided config file
+			for i, f := range backend.Files {
+				for j, p := range f.Paths {
+					if !filepath.IsAbs(p) {
+						backend.Files[i].Paths[j] = filepath.Join(filepath.Dir(configFile), p)
+					}
+				}
+			}
+
+			backend.Backend = &files.FileSearch{
+				FilesBackend: backend.Files,
+			}
+			backends = append(backends, backend)
 		}
-		backends = append(backends, backend)
 	}
+
 	return backends, nil
 }

pkg/files/search.go

@@ -0,0 +1,76 @@
+package files
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/flanksource/flanksource-ui/apm-hub/api/logs"
+)
+
+type FileSearch struct {
+	FilesBackend []logs.FileSearchBackend
+}
+
+func (t *FileSearch) Search(q *logs.SearchParams) (r logs.SearchResults, err error) {
+	var res logs.SearchResults
+
+	for _, b := range t.FilesBackend {
+		if !matchQueryLabels(q.Labels, b.Labels) {
+			continue
+		}
+
+		files, err := readFilesLines(b.Paths)
+		if err != nil {
+			return res, fmt.Errorf("readFilesLines(); %w", err)
+		}
+
+		for _, content := range files {
+			res.Results = append(res.Results, content...)
+		}
+	}
+
+	return res, nil
+}
+
+type logsPerFile map[string][]logs.Result
+
+// readFilesLines will take a list of file paths
+// and then return each lines of those files.
+func readFilesLines(paths []string) (logsPerFile, error) {
+	fileContents := make(logsPerFile, len(paths))
+	for _, path := range paths {
+		fInfo, err := os.Stat(path)
+		if err != nil {
+			return nil, fmt.Errorf("error get file stat. path=%s; %w", path, err)
+		}
+
+		file, err := os.Open(path)
+		if err != nil {
+			return nil, fmt.Errorf("error opening file. path=%s; %w", path, err)
+		}
+
+		scanner := bufio.NewScanner(file)
+		for scanner.Scan() {
+			fileContents[path] = append(fileContents[path], logs.Result{
+				Time: fInfo.ModTime().Format(time.RFC3339),
+				// Labels: , all the records will have the same labels. Is it necessary to add it here?
+				Message: strings.TrimSpace(scanner.Text()),
+			})
+		}
+	}
+
+	return fileContents, nil
+}
+
+func matchQueryLabels(want, have map[string]string) bool {
+	for label, val := range want {
+		if val != have[label] {
+			return false
+		}
+	}
+
+	return true
+}

pkg/files/search_test.go

@@ -0,0 +1,71 @@
+//go:build integration
+
+package files_test
+
+import (
+	"bufio"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/flanksource/flanksource-ui/apm-hub/api/logs"
+	"github.com/flanksource/flanksource-ui/apm-hub/cmd"
+	"github.com/flanksource/flanksource-ui/apm-hub/pkg"
+)
+
+func TestFileSearch(t *testing.T) {
+	confPath := "../../samples/config-file.yaml"
+
+	backend, err := pkg.ParseConfig(nil, confPath)
+	if err != nil {
+		t.Fatal("Fail to parse the config file", err)
+	}
+	logs.GlobalBackends = append(logs.GlobalBackends, backend...)
+
+	sp := logs.SearchParams{
+		Labels: map[string]string{
+			"name": "acmehost",
+			"type": "Nginx",
+		},
+	}
+	b, err := json.Marshal(sp)
+	if err != nil {
+		t.Fatal("Fail to marshal search param")
+	}
+
+	req := httptest.NewRequest(http.MethodPost, "/search", strings.NewReader(string(b)))
+	req.Header.Add("Content-Type", "application/json")
+	rec := httptest.NewRecorder()
+
+	e := cmd.SetupServer(nil)
+	e.ServeHTTP(rec, req)
+
+	var res logs.SearchResults
+	if err := json.NewDecoder(rec.Body).Decode(&res); err != nil {
+		t.Fatal("Failed to decode the search result")
+	}
+
+	nginxLogFile, err := os.Open("../../samples/nginx-access.log")
+	if err != nil {
+		t.Fatal("Fail to read nginx log", err)
+	}
+
+	scanner := bufio.NewScanner(nginxLogFile)
+	var lines []string
+	for scanner.Scan() {
+		lines = append(lines, scanner.Text())
+	}
+
+	if len(res.Results) != len(lines) {
+		t.Fatalf("Expected [%d] lines but got [%d]", len(lines), len(res.Results))
+	}
+
+	for i := range res.Results {
+		if res.Results[i].Message != lines[i] {
+			t.Fatalf("Incorrect line [%d]. Expected %s got %s", i+1, lines[i], res.Results[i].Message)
+		}
+	}
+}

samples/config-file.yaml

@@ -0,0 +1,7 @@
+backends:
+  - file:
+      - labels:
+          name: acmehost
+          type: Nginx
+        path:
+          - nginx-access.log

samples/nginx-access.log

@@ -0,0 +1,3 @@
+127.0.0.1 - - [20/Jan/2023:10:15:30 +0000] "GET /index.html HTTP/1.1" 200 612 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36"
+127.0.0.1 - - [20/Jan/2023:10:15:32 +0000] "GET /about.html HTTP/1.1" 200 754 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36"
+127.0.0.1 - - [20/Jan/2023:10:15:34 +0000] "GET /contact.html HTTP/1.1" 200 834 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.82 Safari/537.36"