admin/kubesphere
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[WIP] logging refactor (#1794)

Browse Source 
* refactor logging

Signed-off-by: huanggze <loganhuang@yunify.com>

* refactor logging

Signed-off-by: huanggze <loganhuang@yunify.com>
This commit is contained in:
Guangzhe Huang
2020-03-02 10:53:43 +08:00
committed by GitHub
parent a9e1183f3c
commit 6c6bfb2677
60 changed files with 1582 additions and 2966 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
pkg/models/log/constants.go
View File

@@ -1,30 +0,0 @@
/*
Copyright 2019 The KubeSphere 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 log
type LogQueryLevel int
const (
QueryLevelCluster LogQueryLevel = iota
QueryLevelWorkspace
QueryLevelNamespace
QueryLevelWorkload
QueryLevelPod
QueryLevelContainer
)

91
pkg/models/log/logcollector.go
View File

@@ -1,91 +0,0 @@
/*
Copyright 2019 The KubeSphere 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 log
import (
"k8s.io/apimachinery/pkg/labels"
"k8s.io/klog"
"kubesphere.io/kubesphere/pkg/constants"
"kubesphere.io/kubesphere/pkg/informers"
"kubesphere.io/kubesphere/pkg/utils/stringutils"
"strconv"
"strings"
"time"
)
// list namespaces that match search conditions
func MatchNamespace(nsFilter []string, nsQuery []string, wsFilter []string, wsQuery []string) (bool, []string) {
nsLister := informers.SharedInformerFactory().Core().V1().Namespaces().Lister()
nsList, err := nsLister.List(labels.Everything())
if err != nil {
klog.Errorf("failed to list namespace, error: %s", err)
return true, nil
}
var namespaces []string
// if no search condition is set on both namespace and workspace,
// then return all namespaces
if nsQuery == nil && nsFilter == nil && wsQuery == nil && wsFilter == nil {
for _, ns := range nsList {
namespaces = append(namespaces, ns.Name)
}
return false, namespaces
}
for _, ns := range nsList {
if stringutils.StringIn(ns.Name, nsFilter) ||
stringutils.StringIn(ns.Annotations[constants.WorkspaceLabelKey], wsFilter) ||
containsIn(ns.Name, nsQuery) ||
containsIn(ns.Annotations[constants.WorkspaceLabelKey], wsQuery) {
namespaces = append(namespaces, ns.Name)
}
}
// if namespaces is equal to nil, indicates no namespace matched
// it causes the query to return no result
return namespaces == nil, namespaces
}
func containsIn(str string, subStrs []string) bool {
for _, sub := range subStrs {
if strings.Contains(str, sub) {
return true
}
}
return false
}
func MakeNamespaceCreationTimeMap(namespaces []string) map[string]string {
namespaceWithCreationTime := make(map[string]string)
nsLister := informers.SharedInformerFactory().Core().V1().Namespaces().Lister()
for _, item := range namespaces {
ns, err := nsLister.Get(item)
if err != nil {
// the ns doesn't exist
continue
}
namespaceWithCreationTime[ns.Name] = strconv.FormatInt(ns.CreationTimestamp.UnixNano()/int64(time.Millisecond), 10)
}
return namespaceWithCreationTime
}

368
pkg/models/log/logcrd.go
View File

@@ -1,368 +0,0 @@
/*
Copyright 2018 The KubeSphere 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 log
import (
_ "github.com/go-sql-driver/mysql"
"github.com/google/uuid"
"github.com/json-iterator/go"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/klog"
"kubesphere.io/kubesphere/pkg/apis/logging/v1alpha2"
"kubesphere.io/kubesphere/pkg/informers"
"net/http"
"strings"
"time"
)
var jsonIter = jsoniter.ConfigCompatibleWithStandardLibrary
const (
ConfigMapName = "fluent-bit-output-config"
ConfigMapData = "outputs"
LoggingNamespace = "kubesphere-logging-system"
)
func createCRDClientSet() (*rest.RESTClient, *runtime.Scheme, error) {
config, err := fb.GetClientConfig("")
if err != nil {
//panic(err.Error())
return nil, nil, err
}
// Create a new clientset which include our CRD schema
return fb.NewFluentbitCRDClient(config)
}
func FluentbitOutputsQuery() *v1alpha2.FluentbitOutputsResult {
var result v1alpha2.FluentbitOutputsResult
outputs, err := GetFluentbitOutputFromConfigMap()
if err != nil {
result.Status = http.StatusInternalServerError
result.Error = err.Error()
return &result
}
result.Outputs = outputs
result.Status = http.StatusOK
return &result
}
func FluentbitOutputInsert(output v1alpha2.OutputPlugin) *v1alpha2.FluentbitOutputsResult {
var result v1alpha2.FluentbitOutputsResult
// 1. Update ConfigMap
var outputs []v1alpha2.OutputPlugin
outputs, err := GetFluentbitOutputFromConfigMap()
if err != nil {
// If the ConfigMap doesn't exist, a new one will be created later
klog.Errorln(err)
}
// When adding a new output for the first time, one should always set it enabled
output.Enable = true
output.Id = uuid.New().String()
output.Updatetime = time.Now()
outputs = append(outputs, output)
err = updateFluentbitOutputConfigMap(outputs)
if err != nil {
result.Status = http.StatusInternalServerError
result.Error = err.Error()
return &result
}
// 2. Keep CRD in inline with ConfigMap
err = syncFluentbitCRDOutputWithConfigMap(outputs)
if err != nil {
result.Status = http.StatusInternalServerError
result.Error = err.Error()
return &result
}
result.Status = http.StatusOK
return &result
}
func FluentbitOutputUpdate(output v1alpha2.OutputPlugin, id string) *v1alpha2.FluentbitOutputsResult {
var result v1alpha2.FluentbitOutputsResult
// 1. Update ConfigMap
var outputs []v1alpha2.OutputPlugin
outputs, err := GetFluentbitOutputFromConfigMap()
if err != nil {
// If the ConfigMap doesn't exist, a new one will be created later
klog.Errorln(err)
}
index := 0
for _, output := range outputs {
if output.Id == id {
break
}
index++
}
if index >= len(outputs) {
result.Status = http.StatusNotFound
result.Error = "The output plugin to update doesn't exist. Please check the output id you provide."
return &result
}
output.Updatetime = time.Now()
outputs = append(append(outputs[:index], outputs[index+1:]...), output)
err = updateFluentbitOutputConfigMap(outputs)
if err != nil {
result.Status = http.StatusInternalServerError
result.Error = err.Error()
return &result
}
// 2. Keep CRD in inline with ConfigMap
err = syncFluentbitCRDOutputWithConfigMap(outputs)
if err != nil {
result.Status = http.StatusInternalServerError
result.Error = err.Error()
return &result
}
result.Status = http.StatusOK
return &result
}
func FluentbitOutputDelete(id string) *v1alpha2.FluentbitOutputsResult {
var result v1alpha2.FluentbitOutputsResult
// 1. Update ConfigMap
// If the ConfigMap doesn't exist, a new one will be created
outputs, _ := GetFluentbitOutputFromConfigMap()
index := 0
for _, output := range outputs {
if output.Id == id {
break
}
index++
}
if index >= len(outputs) {
result.Status = http.StatusNotFound
result.Error = "The output plugin to delete doesn't exist. Please check the output id you provide."
return &result
}
outputs = append(outputs[:index], outputs[index+1:]...)
err := updateFluentbitOutputConfigMap(outputs)
if err != nil {
result.Status = http.StatusInternalServerError
result.Error = err.Error()
return &result
}
// 2. Keep CRD in inline with DB
err = syncFluentbitCRDOutputWithConfigMap(outputs)
if err != nil {
result.Status = http.StatusInternalServerError
result.Error = err.Error()
return &result
}
result.Status = http.StatusOK
return &result
}
func GetFluentbitOutputFromConfigMap() ([]v1alpha2.OutputPlugin, error) {
configMap, err := informers.SharedInformerFactory().Core().V1().ConfigMaps().Lister().ConfigMaps(LoggingNamespace).Get(ConfigMapName)
if err != nil {
return nil, err
}
data := configMap.Data[ConfigMapData]
var outputs []fb.OutputPlugin
if err = jsonIter.UnmarshalFromString(data, &outputs); err != nil {
return nil, err
}
return outputs, nil
}
func updateFluentbitOutputConfigMap(outputs []fb.OutputPlugin) error {
var data string
data, err := jsonIter.MarshalToString(outputs)
if err != nil {
klog.Errorln(err)
return err
}
// Update the ConfigMap
config, err := rest.InClusterConfig()
if err != nil {
klog.Errorln(err)
return err
}
// Creates the clientset
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
klog.Errorln(err)
return err
}
configMapClient := clientset.CoreV1().ConfigMaps(LoggingNamespace)
configMap, err := configMapClient.Get(ConfigMapName, metav1.GetOptions{})
if err != nil {
// If the ConfigMap doesn't exist, create a new one
newConfigMap := &corev1.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
Name: ConfigMapName,
},
Data: map[string]string{ConfigMapData: data},
}
_, err = configMapClient.Create(newConfigMap)
if err != nil {
klog.Errorln(err)
return err
}
} else {
// update
configMap.Data = map[string]string{ConfigMapData: data}
_, err = configMapClient.Update(configMap)
if err != nil {
klog.Errorln(err)
return err
}
}
return nil
}
func syncFluentbitCRDOutputWithConfigMap(outputs []v1alpha2.OutputPlugin) error {
var enabledOutputs []v1alpha2.Plugin
for _, output := range outputs {
if output.Enable {
enabledOutputs = append(enabledOutputs, v1alpha2.Plugin{Type: output.Type, Name: output.Name, Parameters: output.Parameters})
}
}
// Empty output is not allowed, must specify a null-type output
if len(enabledOutputs) == 0 {
enabledOutputs = []v1alpha2.Plugin{
{
Type: "fluentbit_output",
Name: "fluentbit-output-null",
Parameters: []v1alpha2.Parameter{
{
Name: "Name",
Value: "null",
},
{
Name: "Match",
Value: "*",
},
},
},
}
}
crdcs, scheme, err := createCRDClientSet()
if err != nil {
klog.Errorln(err)
return err
}
// Create a CRD client interface
crdclient := v1alpha2.CrdClient(crdcs, scheme, LoggingNamespace)
fluentbit, err := crdclient.Get("fluent-bit")
if err != nil {
klog.Errorln(err)
return err
}
fluentbit.Spec.Output = enabledOutputs
_, err = crdclient.Update("fluent-bit", fluentbit)
if err != nil {
klog.Errorln(err)
return err
}
return nil
}
// Parse es host, port and index
func ParseEsOutputParams(params []v1alpha2.Parameter) *v1alpha2.Config {
var (
isEsFound bool
host = "127.0.0.1"
port = "9200"
index = "logstash"
logstashFormat string
logstashPrefix string
)
for _, param := range params {
switch param.Name {
case "Name":
if param.Value == "es" {
isEsFound = true
}
case "Host":
host = param.Value
case "Port":
port = param.Value
case "Index":
index = param.Value
case "Logstash_Format":
logstashFormat = strings.ToLower(param.Value)
case "Logstash_Prefix":
logstashPrefix = param.Value
}
}
if !isEsFound {
return nil
}
// If Logstash_Format is On/True, ignore Index
if logstashFormat == "on" || logstashFormat == "true" {
if logstashPrefix != "" {
index = logstashPrefix
} else {
index = "logstash"
}
}
return &v1alpha2.Config{Host: host, Port: port, Index: index}
}

19
pkg/models/log/types.go
View File

@@ -1,19 +0,0 @@
/*
Copyright 2019 The KubeSphere 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 log

41
pkg/models/logging/logging.go Normal file
View File

@@ -0,0 +1,41 @@
package logging
import (
"io"
"kubesphere.io/kubesphere/pkg/api/logging/v1alpha2"
"kubesphere.io/kubesphere/pkg/simple/client/logging"
)
type LoggingOperator interface {
GetCurrentStats(sf logging.SearchFilter) (v1alpha2.APIResponse, error)
CountLogsByInterval(sf logging.SearchFilter, interval string) (v1alpha2.APIResponse, error)
ExportLogs(sf logging.SearchFilter, w io.Writer) error
SearchLogs(sf logging.SearchFilter, from, size int64, order string) (v1alpha2.APIResponse, error)
}
type loggingOperator struct {
c logging.Interface
}
func NewLoggingOperator(client logging.Interface) LoggingOperator {
return &loggingOperator{client}
}
func (l loggingOperator) GetCurrentStats(sf logging.SearchFilter) (v1alpha2.APIResponse, error) {
res, err := l.c.GetCurrentStats(sf)
return v1alpha2.APIResponse{Statistics: &res}, err
}
func (l loggingOperator) CountLogsByInterval(sf logging.SearchFilter, interval string) (v1alpha2.APIResponse, error) {
res, err := l.c.CountLogsByInterval(sf, interval)
return v1alpha2.APIResponse{Histogram: &res}, err
}
func (l loggingOperator) ExportLogs(sf logging.SearchFilter, w io.Writer) error {
return l.c.ExportLogs(sf, w)
}
func (l loggingOperator) SearchLogs(sf logging.SearchFilter, from, size int64, order string) (v1alpha2.APIResponse, error) {
res, err := l.c.SearchLogs(sf, from, size, order)
return v1alpha2.APIResponse{Logs: &res}, err
}