Automated cherry pick of #1841: feature: log query improvement

modules/extensions/loghub/index/query/clients.go

@@ -25,7 +25,8 @@ import (
 	"github.com/recallsong/go-utils/encoding/jsonx"
 	"github.com/recallsong/go-utils/reflectx"
 
-	"github.com/erda-project/erda/modules/msp/instance/db"
+	"github.com/erda-project/erda/modules/extensions/loghub/index/query/db"
+	mspDb "github.com/erda-project/erda/modules/msp/instance/db"
 	"github.com/erda-project/erda/pkg/http/httpclient"
 )
 
@@ -38,9 +39,11 @@ const (
 // ESClient .
 type ESClient struct {
 	*elastic.Client
+	Cluster    string
 	URLs       string
 	LogVersion string
 	Indices    []string
+	Entrys     []*IndexEntry
 }
 
 func (c *ESClient) printSearchSource(searchSource *elastic.SearchSource) (string, error) {
@@ -54,6 +57,14 @@ func (c *ESClient) printSearchSource(searchSource *elastic.SearchSource) (string
 	return body, nil
 }
 
+func (p *provider) getAllESClients() []*ESClient {
+	list, err := p.db.LogDeployment.List()
+	if err != nil {
+		return nil
+	}
+	return p.getESClientsFromLogAnalyticsByLogDeployment("", list...)
+}
+
 func (p *provider) getESClients(orgID int64, req *LogRequest) []*ESClient {
 	if len(req.ClusterName) > 0 || len(req.Addon) > 0 {
 		if len(req.ClusterName) <= 0 || len(req.Addon) <= 0 {
@@ -110,13 +121,24 @@ func (p *provider) getESClientsFromLogAnalyticsByCluster(orgID int64, addon stri
 	if err != nil {
 		return nil
 	}
+	return p.getESClientsFromLogAnalyticsByLogDeployment(addon, list...)
+}
+
+func (p *provider) getESClientsFromLogAnalyticsByLogDeployment(addon string, logDeployments ...*db.LogDeployment) []*ESClient {
 	type ESConfig struct {
 		Security bool   `json:"securityEnable"`
 		Username string `json:"securityUsername"`
 		Password string `json:"securityPassword"`
 	}
+	var indices map[string]map[string][]*IndexEntry
+	if p.C.IndexPreload {
+		v := p.indices.Load()
+		if v != nil {
+			indices = v.(map[string]map[string][]*IndexEntry)
+		}
+	}
 	var clients []*ESClient
-	for _, d := range list {
+	for _, d := range logDeployments {
 		if len(d.ESURL) <= 0 {
 			continue
 		}
@@ -137,7 +159,7 @@ func (p *provider) getESClientsFromLogAnalyticsByCluster(orgID int64, addon stri
 				}
 				for _, instance := range sameGroupLogInstances {
 					logKey := instance.LogKey
-					if instance.LogType == string(db.LogTypeLogService) {
+					if instance.LogType == string(mspDb.LogTypeLogService) {
 						var instanceConfig = struct {
 							MspEnvID string `json:"MSP_ENV_ID"`
 						}{}
@@ -176,7 +198,7 @@ func (p *provider) getESClientsFromLogAnalyticsByCluster(orgID int64, addon stri
 		}
 
 		orgId := d.OrgID
-		if d.LogType == string(db.LogTypeLogAnalytics) {
+		if d.LogType == string(mspDb.LogTypeLogAnalytics) {
 			// omit the orgId alias, if deployed by log-analytics addon，specially for old versions, there's no orgId alias
 			orgId = ""
 		}
@@ -187,16 +209,27 @@ func (p *provider) getESClientsFromLogAnalyticsByCluster(orgID int64, addon stri
 			continue
 		}
 		d.CollectorURL = strings.TrimSpace(d.CollectorURL)
-		if len(d.CollectorURL) > 0 || d.LogType == string(db.LogTypeLogService) {
-			clients = append(clients, &ESClient{
+		if len(d.CollectorURL) > 0 || d.LogType == string(mspDb.LogTypeLogService) {
+			c := &ESClient{
 				Client:     client,
+				Cluster:    d.ClusterName,
 				LogVersion: LogVersion2,
 				URLs:       d.ESURL,
 				Indices:    getLogIndices("rlogs-", orgId, addons...),
-			})
+			}
+			clients = append(clients, c)
+
+			if p.C.IndexPreload && indices != nil && len(addons) > 0 {
+				if indexAddons, ok := indices[d.ClusterName]; ok {
+					for _, addon := range addons {
+						c.Entrys = append(c.Entrys, indexAddons[addon]...)
+					}
+				}
+			}
 		} else {
 			clients = append(clients, &ESClient{
 				Client:     client,
+				Cluster:    d.ClusterName,
 				LogVersion: LogVersion1,
 				URLs:       d.ESURL,
 				Indices:    getLogIndices("spotlogs-", orgId, addons...),

modules/extensions/loghub/index/query/clients_test.go

@@ -15,6 +15,7 @@
 package query
 
 import (
+	"fmt"
 	"reflect"
 	"testing"
 
@@ -27,6 +28,17 @@ import (
 	db2 "github.com/erda-project/erda/modules/msp/instance/db"
 )
 
+func TestNewESClient(t *testing.T) {
+	c := ESClient{
+		Cluster: "cluster-1",
+		Entrys:  []*IndexEntry{},
+	}
+
+	if len(c.Cluster) == 0 {
+		t.Log("hennnnnnnn...")
+	}
+}
+
 func TestGetLogIndices_WithNoneEmptyOrgId_Should_Return_Indices_With_OrgAlias(t *testing.T) {
 	result := getLogIndices("rlogs-", "1")
 	if len(result) == 0 {
@@ -77,3 +89,117 @@ func TestGetESClientsFromLogAnalyticsByCluster_Should_Success(t *testing.T) {
 
 	p.getESClientsFromLogAnalyticsByCluster(1, "addon-1", "cluster-1")
 }
+
+func TestGetAllESClients_WithErrorAccessDb_Should_Return_Nil(t *testing.T) {
+	p := &provider{
+		db: &db.DB{
+			LogDeployment: db.LogDeploymentDB{},
+		},
+	}
+
+	defer monkey.UnpatchInstanceMethod(reflect.TypeOf(&p.db.LogDeployment), "List")
+	monkey.PatchInstanceMethod(reflect.TypeOf(&p.db.LogDeployment), "List", func(_ *db.LogDeploymentDB) ([]*db.LogDeployment, error) {
+		return nil, fmt.Errorf("boooooo!")
+	})
+
+	clients := p.getAllESClients()
+	if clients != nil {
+		t.Errorf("should return nil when fail to access logDeployment")
+	}
+}
+
+func TestGetAllESClients_On_ExistsLogDeployment_Should_Return_None_Empty_Clients(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	p := provider{
+		L:     NewMockLogger(ctrl),
+		C:     &config{},
+		mysql: &gorm.DB{},
+		db: &db.DB{
+			LogDeployment:        db.LogDeploymentDB{},
+			LogServiceInstanceDB: db.LogServiceInstanceDB{},
+			LogInstanceDB:        db.LogInstanceDB{},
+		},
+		bdl:        bundle.New(),
+		timeRanges: make(map[string]map[string]*timeRange),
+		reload:     make(chan struct{}),
+	}
+
+	defer monkey.UnpatchInstanceMethod(reflect.TypeOf(&p.db.LogDeployment), "List")
+	monkey.PatchInstanceMethod(reflect.TypeOf(&p.db.LogDeployment), "List", func(_ *db.LogDeploymentDB) ([]*db.LogDeployment, error) {
+		return []*db.LogDeployment{
+			&db.LogDeployment{
+				ClusterName:  "cluster-1",
+				ClusterType:  0,
+				ESURL:        "http://localhost:9200",
+				ESConfig:     "{}",
+				CollectorURL: "http://collector:7096",
+			},
+		}, nil
+	})
+
+	defer monkey.UnpatchInstanceMethod(reflect.TypeOf(&p.db.LogDeployment), "QueryByOrgIDAndClusters")
+	monkey.PatchInstanceMethod(reflect.TypeOf(&p.db.LogDeployment), "QueryByOrgIDAndClusters", func(_ *db.LogDeploymentDB, orgID int64, clusters ...string) ([]*db.LogDeployment, error) {
+		return []*db.LogDeployment{
+			{LogType: string(db2.LogTypeLogService), ESURL: "http://localhost:9200"},
+		}, nil
+	})
+
+	defer monkey.UnpatchInstanceMethod(reflect.TypeOf(&p.db.LogInstanceDB), "GetByLogKey")
+	monkey.PatchInstanceMethod(reflect.TypeOf(&p.db.LogInstanceDB), "GetByLogKey", func(_ *db.LogInstanceDB, logKey string) (*db.LogInstance, error) {
+		return &db.LogInstance{LogType: string(db2.LogTypeLogAnalytics), LogKey: "logKey-1", Config: `{"MSP_ENV_ID":"msp_env_id_1"}`}, nil
+	})
+
+	defer monkey.UnpatchInstanceMethod(reflect.TypeOf(&p.db.LogInstanceDB), "GetListByClusterAndProjectIdAndWorkspace")
+	monkey.PatchInstanceMethod(reflect.TypeOf(&p.db.LogInstanceDB), "GetListByClusterAndProjectIdAndWorkspace", func(_ *db.LogInstanceDB, clusterName, projectId, workspace string) ([]db.LogInstance, error) {
+		return []db.LogInstance{
+			{LogType: string(db2.LogTypeLogService), LogKey: "logKey-3", Config: `{"MSP_ENV_ID":"msp_env_id_1"}`},
+			{LogType: string(db2.LogTypeLogService), LogKey: "logKey-2", Config: `{"MSP_ENV_ID":"msp_env_id_1"}`},
+			{LogType: string(db2.LogTypeLogAnalytics), LogKey: "logKey-1", Config: `{"MSP_ENV_ID":"msp_env_id_1"}`},
+		}, nil
+	})
+
+	clients := p.getAllESClients()
+	if len(clients) == 0 {
+		t.Errorf("should return non-empty ESClients list when exists logDeployment")
+	}
+}
+
+/*
+func TestGetESClientsFromLogAnalyticsByLogDeployment_On_Preload_Enabled_Should_Try_Fill_ESClient_Entrys(t *testing.T) {
+	p := &provider{
+		db: &db.DB{
+			LogDeployment: db.LogDeploymentDB{},
+		},
+		C: &config{
+			IndexPreload: true,
+		},
+		timeRanges: make(map[string]map[string]*timeRange),
+		reload:     make(chan struct{}),
+	}
+	p.indices.Store(map[string]map[string][]*IndexEntry{
+		"cluster_1": map[string][]*IndexEntry{
+			"addon_1": []*IndexEntry{
+				&IndexEntry{Index: "rlogs-addon_1-2020.34-000001",
+					Name: "addon_1",
+				},
+			},
+		},
+	})
+
+	logDeployments := []*db.LogDeployment{
+		&db.LogDeployment{
+			ClusterName:  "cluster_1",
+			ClusterType:  0,
+			ESURL:        "http://localhost:9200",
+			ESConfig:     "{}",
+			CollectorURL: "http://collector:7096",
+		},
+	}
+
+	clients := p.getESClientsFromLogAnalyticsByLogDeployment("addon_1", logDeployments...)
+	if len(clients) == 0 || len(clients[0].Entrys) == 0 {
+		t.Errorf("ESClient.Entrys should not empty when preload matched")
+	}
+}
+*/

modules/extensions/loghub/index/query/db/log_deployment.go

@@ -77,3 +77,13 @@ func (db *LogDeploymentDB) QueryByOrgIDAndClusters(orgID int64, clusters ...stri
 	}
 	return list, nil
 }
+
+// List .
+func (db *LogDeploymentDB) List() ([]*LogDeployment, error) {
+	var list []*LogDeployment
+	if err := db.Table(LogDeploymentTable).
+		Find(&list).Error; err != nil {
+		return nil, err
+	}
+	return list, nil
+}