Merge pull request #31 from clidey/hk/feature/elasticsearch

[Feature] Add support for Elastic Search

Author: hkdeman

core/go.mod

@@ -4,6 +4,8 @@ go 1.22.1
 
 require (
 	github.com/99designs/gqlgen v0.17.48
+	github.com/elastic/go-elasticsearch v0.0.0
+	github.com/elastic/go-elasticsearch/v8 v8.14.0
 	github.com/go-chi/chi/v5 v5.0.12
 	github.com/go-chi/cors v1.2.1
 	github.com/go-redis/redis/v8 v8.11.5
@@ -22,6 +24,9 @@ require (
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.4 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/elastic/elastic-transport-go/v8 v8.6.0 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-sql-driver/mysql v1.7.0 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
@@ -45,6 +50,9 @@ require (
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/xrash/smetrics v0.0.0-20240312152122-5f08fbb34913 // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
+	go.opentelemetry.io/otel v1.24.0 // indirect
+	go.opentelemetry.io/otel/metric v1.24.0 // indirect
+	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	golang.org/x/crypto v0.22.0 // indirect
 	golang.org/x/mod v0.17.0 // indirect
 	golang.org/x/sync v0.7.0 // indirect

core/go.sum

@@ -22,12 +22,23 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/r
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48 h1:fRzb/w+pyskVMQ+UbP35JkH8yB7MYb4q/qhBarqZE6g=
 github.com/dgryski/trifles v0.0.0-20200323201526-dd97f9abfb48/go.mod h1:if7Fbed8SFyPtHLHbg49SI7NAdJiC5WIA09pe59rfAA=
+github.com/elastic/elastic-transport-go/v8 v8.6.0 h1:Y2S/FBjx1LlCv5m6pWAF2kDJAHoSjSRSJCApolgfthA=
+github.com/elastic/elastic-transport-go/v8 v8.6.0/go.mod h1:YLHer5cj0csTzNFXoNQ8qhtGY1GTvSqPnKWKaqQE3Hk=
+github.com/elastic/go-elasticsearch v0.0.0 h1:Pd5fqOuBxKxv83b0+xOAJDAkziWYwFinWnBO0y+TZaA=
+github.com/elastic/go-elasticsearch v0.0.0/go.mod h1:TkBSJBuTyFdBnrNqoPc54FN0vKf5c04IdM4zuStJ7xg=
+github.com/elastic/go-elasticsearch/v8 v8.14.0 h1:1ywU8WFReLLcxE1WJqii3hTtbPUE2hc38ZK/j4mMFow=
+github.com/elastic/go-elasticsearch/v8 v8.14.0/go.mod h1:WRvnlGkSuZyp83M2U8El/LGXpCjYLrvlkSgkAH4O5I4=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/go-chi/chi/v5 v5.0.12 h1:9euLV5sTrTNTRUU9POmDUvfxyj6LAABLUcEWO+JJb4s=
 github.com/go-chi/chi/v5 v5.0.12/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
 github.com/go-chi/cors v1.2.1 h1:xEC8UT3Rlp2QuWNEr4Fs/c2EAGVKBwy/1vHx3bppil4=
 github.com/go-chi/cors v1.2.1/go.mod h1:sSbTewc+6wYHBBCW7ytsFSn836hqM7JxpglAy2Vzc58=
+github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
+github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
 github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
 github.com/go-sql-driver/mysql v1.7.0 h1:ueSltNNllEqE3qcWBTD0iQd3IpL/6U+mJxLkazJ7YPc=
@@ -104,6 +115,14 @@ github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7Jul
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.mongodb.org/mongo-driver v1.16.0 h1:tpRsfBJMROVHKpdGyc1BBEzzjDUWjItxbVSZ8Ls4BQ4=
 go.mongodb.org/mongo-driver v1.16.0/go.mod h1:oB6AhJQvFQL4LEHyXi6aJzQJtBiTQHiAd83l0GdFaiw=
+go.opentelemetry.io/otel v1.24.0 h1:0LAOdjNmQeSTzGBzduGe/rU4tZhMwL5rWgtp9Ku5Jfo=
+go.opentelemetry.io/otel v1.24.0/go.mod h1:W7b9Ozg4nkF5tWI5zsXkaKKDjdVjpD4oAt9Qi/MArHo=
+go.opentelemetry.io/otel/metric v1.24.0 h1:6EhoGWWK28x1fbpA4tYTOWBkPefTDQnb8WSGXlc88kI=
+go.opentelemetry.io/otel/metric v1.24.0/go.mod h1:VYhLe1rFfxuTXLgj4CBiyz+9WYBA8pNGJgDcSFRKBco=
+go.opentelemetry.io/otel/sdk v1.21.0 h1:FTt8qirL1EysG6sTQRZ5TokkU8d0ugCj8htOgThZXQ8=
+go.opentelemetry.io/otel/sdk v1.21.0/go.mod h1:Nna6Yv7PWTdgJHVRD9hIYywQBRx7pbox6nwBnZIxl/E=
+go.opentelemetry.io/otel/trace v1.24.0 h1:CsKnnL4dUAr/0llH9FKuc698G04IrpWV0MQA/Y1YELI=
+go.opentelemetry.io/otel/trace v1.24.0/go.mod h1:HPc3Xr/cOApsBI154IU0OI0HJexz+aw5uPdbs3UCjNU=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.22.0 h1:g1v0xeRhjcugydODzvb3mEM9SQ0HGp9s/nh3COQ/C30=

core/graph/model/models_gen.go

@@ -8,6 +8,7 @@ enum DatabaseType {
   Sqlite3,
   MongoDB,
   Redis,
+  ElasticSearch,
 }
 
 type Column {

core/graph/schema.graphqls

@@ -5,11 +5,12 @@ import "github.com/clidey/whodb/core/graph/model"
 type DatabaseType string
 
 const (
-	DatabaseType_Postgres = "Postgres"
-	DatabaseType_MySQL    = "MySQL"
-	DatabaseType_Sqlite3  = "Sqlite3"
-	DatabaseType_MongoDB  = "MongoDB"
-	DatabaseType_Redis    = "Redis"
+	DatabaseType_Postgres      = "Postgres"
+	DatabaseType_MySQL         = "MySQL"
+	DatabaseType_Sqlite3       = "Sqlite3"
+	DatabaseType_MongoDB       = "MongoDB"
+	DatabaseType_Redis         = "Redis"
+	DatabaseType_ElasticSearch = "ElasticSearch"
 )
 
 type Engine struct {

core/src/engine/engine.go

@@ -0,0 +1,42 @@
+package elasticsearch
+
+import (
+	"fmt"
+
+	"github.com/clidey/whodb/core/src/common"
+	"github.com/clidey/whodb/core/src/engine"
+	"github.com/elastic/go-elasticsearch/v8"
+)
+
+func DB(config *engine.PluginConfig) (*elasticsearch.Client, error) {
+	var addresses []string
+	port := common.GetRecordValueOrDefault(config.Credentials.Advanced, "Port", "9200")
+	sslMode := common.GetRecordValueOrDefault(config.Credentials.Advanced, "SSL Mode", "disable")
+	if sslMode == "enable" {
+		addresses = []string{
+			fmt.Sprintf("https://%s:%s", config.Credentials.Hostname, port),
+		}
+	} else {
+		addresses = []string{
+			fmt.Sprintf("http://%s:%s", config.Credentials.Hostname, port),
+		}
+	}
+
+	cfg := elasticsearch.Config{
+		Addresses: addresses,
+		Username:  config.Credentials.Username,
+		Password:  config.Credentials.Password,
+	}
+
+	client, err := elasticsearch.NewClient(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := client.Info()
+	if err != nil || res.IsError() {
+		return nil, fmt.Errorf("error pinging Elasticsearch: %v", err)
+	}
+
+	return client, nil
+}

core/src/plugins/elasticsearch/db.go

@@ -0,0 +1,159 @@
+package elasticsearch
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/clidey/whodb/core/src/engine"
+)
+
+type ElasticSearchPlugin struct{}
+
+func (p *ElasticSearchPlugin) IsAvailable(config *engine.PluginConfig) bool {
+	client, err := DB(config)
+	if err != nil {
+		return false
+	}
+	res, err := client.Info()
+	if err != nil || res.IsError() {
+		return false
+	}
+	return true
+}
+
+func (p *ElasticSearchPlugin) GetDatabases() ([]string, error) {
+	return nil, errors.ErrUnsupported
+}
+
+func (p *ElasticSearchPlugin) GetSchema(config *engine.PluginConfig) ([]string, error) {
+	return nil, errors.ErrUnsupported
+}
+
+func (p *ElasticSearchPlugin) GetStorageUnits(config *engine.PluginConfig, database string) ([]engine.StorageUnit, error) {
+	client, err := DB(config)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := client.Indices.Stats()
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		return nil, fmt.Errorf("error getting stats for indices: %s", res.String())
+	}
+
+	var stats map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&stats); err != nil {
+		return nil, err
+	}
+
+	indicesStats := stats["indices"].(map[string]interface{})
+	storageUnits := make([]engine.StorageUnit, 0, len(indicesStats))
+
+	for indexName, indexStatsInterface := range indicesStats {
+		indexStats := indexStatsInterface.(map[string]interface{})
+		primaries := indexStats["primaries"].(map[string]interface{})
+		docs := primaries["docs"].(map[string]interface{})
+		store := primaries["store"].(map[string]interface{})
+
+		storageUnit := engine.StorageUnit{
+			Name: indexName,
+			Attributes: []engine.Record{
+				{Key: "Storage Size", Value: fmt.Sprintf("%v", store["size_in_bytes"])},
+				{Key: "Count", Value: fmt.Sprintf("%v", docs["count"])},
+			},
+		}
+		storageUnits = append(storageUnits, storageUnit)
+	}
+
+	return storageUnits, nil
+}
+
+func (p *ElasticSearchPlugin) GetRows(config *engine.PluginConfig, database, collection, filter string, pageSize, pageOffset int) (*engine.GetRowsResult, error) {
+	client, err := DB(config)
+	if err != nil {
+		return nil, err
+	}
+
+	var elasticSearchConditions map[string]interface{}
+	if len(filter) > 0 {
+		if err := json.Unmarshal([]byte(filter), &elasticSearchConditions); err != nil {
+			return nil, fmt.Errorf("invalid filter format: %v", err)
+		}
+	}
+
+	query := map[string]interface{}{
+		"from": pageOffset,
+		"size": pageSize,
+	}
+
+	for key, value := range elasticSearchConditions {
+		query[key] = value
+	}
+
+	var buf bytes.Buffer
+	if err := json.NewEncoder(&buf).Encode(query); err != nil {
+		return nil, err
+	}
+
+	res, err := client.Search(
+		client.Search.WithContext(context.Background()),
+		client.Search.WithIndex(collection),
+		client.Search.WithBody(&buf),
+		client.Search.WithTrackTotalHits(true),
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+
+	if res.IsError() {
+		return nil, fmt.Errorf("error searching documents: %s", res.String())
+	}
+
+	var searchResult map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&searchResult); err != nil {
+		return nil, err
+	}
+
+	hits := searchResult["hits"].(map[string]interface{})["hits"].([]interface{})
+	result := &engine.GetRowsResult{
+		Columns: []engine.Column{
+			{Name: "document", Type: "Document"},
+		},
+		Rows: [][]string{},
+	}
+
+	for _, hit := range hits {
+		hitMap := hit.(map[string]interface{})
+		source := hitMap["_source"]
+		id := hitMap["_id"]
+		document := map[string]interface{}{}
+		document["_id"] = id
+		document["source"] = source
+		jsonBytes, err := json.Marshal(document)
+		if err != nil {
+			return nil, err
+		}
+		result.Rows = append(result.Rows, []string{string(jsonBytes)})
+	}
+
+	return result, nil
+}
+
+func (p *ElasticSearchPlugin) RawExecute(config *engine.PluginConfig, query string) (*engine.GetRowsResult, error) {
+	return nil, errors.New("unsupported operation")
+}
+
+func NewElasticSearchPlugin() *engine.Plugin {
+	return &engine.Plugin{
+		Type:            engine.DatabaseType_ElasticSearch,
+		PluginFunctions: &ElasticSearchPlugin{},
+	}
+}