Merge pull request #2069 from junotx/master

add events search apis

cmd/ks-apiserver/app/options/options.go

@@ -13,6 +13,7 @@ import (
 	genericoptions "kubesphere.io/kubesphere/pkg/server/options"
 	"kubesphere.io/kubesphere/pkg/simple/client/cache"
 	"kubesphere.io/kubesphere/pkg/simple/client/devops/jenkins"
+	eventsclient "kubesphere.io/kubesphere/pkg/simple/client/events/elasticsearch"
 	"kubesphere.io/kubesphere/pkg/simple/client/k8s"
 	"kubesphere.io/kubesphere/pkg/simple/client/ldap"
 	esclient "kubesphere.io/kubesphere/pkg/simple/client/logging/elasticsearch"
@@ -54,6 +55,7 @@ func NewServerRunOptions() *ServerRunOptions {
 			RedisOptions:          cache.NewRedisOptions(),
 			AuthenticationOptions: authoptions.NewAuthenticateOptions(),
 			MultiClusterOptions:   multicluster.NewOptions(),
+			EventsOptions:         eventsclient.NewElasticSearchOptions(),
 		},
 	}
 
@@ -78,6 +80,7 @@ func (s *ServerRunOptions) Flags() (fss cliflag.NamedFlagSets) {
 	s.MonitoringOptions.AddFlags(fss.FlagSet("monitoring"), s.MonitoringOptions)
 	s.LoggingOptions.AddFlags(fss.FlagSet("logging"), s.LoggingOptions)
 	s.MultiClusterOptions.AddFlags(fss.FlagSet("multicluster"), s.MultiClusterOptions)
+	s.EventsOptions.AddFlags(fss.FlagSet("events"), s.EventsOptions)
 
 	fs = fss.FlagSet("klog")
 	local := flag.NewFlagSet("klog", flag.ExitOnError)
@@ -177,6 +180,14 @@ func (s *ServerRunOptions) NewAPIServer(stopCh <-chan struct{}) (*apiserver.APIS
 		}
 	}
 
+	if s.EventsOptions.Host != "" {
+		eventsClient, err := eventsclient.NewClient(s.EventsOptions)
+		if err != nil {
+			return nil, err
+		}
+		apiServer.EventsClient = eventsClient
+	}
+
 	if s.OpenPitrixOptions != nil {
 		opClient, err := openpitrix.NewClient(s.OpenPitrixOptions)
 		if err != nil {

cmd/ks-apiserver/app/options/validation.go

@@ -16,6 +16,7 @@ func (s *ServerRunOptions) Validate() []error {
 	errors = append(errors, s.NetworkOptions.Validate()...)
 	errors = append(errors, s.LoggingOptions.Validate()...)
 	errors = append(errors, s.AuthorizationOptions.Validate()...)
+	errors = append(errors, s.EventsOptions.Validate()...)
 
 	return errors
 }

pkg/api/events/v1alpha1/types.go

+package v1alpha1
+
+import (
+	"github.com/emicklei/go-restful"
+	"kubesphere.io/kubesphere/pkg/simple/client/events"
+	"strconv"
+	"time"
+)
+
+type APIResponse struct {
+	Events     *events.Events     `json:"query,omitempty" description:"query results"`
+	Statistics *events.Statistics `json:"statistics,omitempty" description:"statistics results"`
+	Histogram  *events.Histogram  `json:"histogram,omitempty" description:"histogram results"`
+}
+
+type Query struct {
+	Operation                     string `json:"operation,omitempty"`
+	WorkspaceFilter               string `json:"workspace_filter,omitempty"`
+	WorkspaceSearch               string `json:"workspace_search,omitempty"`
+	InvolvedObjectNamespaceFilter string `json:"involved_object_namespace_filter,omitempty"`
+	InvolvedObjectNamespaceSearch string `json:"involved_object_namespace_search,omitempty"`
+	InvolvedObjectNameFilter      string `json:"involved_object_name_filter,omitempty"`
+	InvolvedObjectNameSearch      string `json:"involved_object_name_search,omitempty"`
+	InvolvedObjectKindFilter      string `json:"involved_object_kind_filter,omitempty"`
+	ReasonFilter                  string `json:"reason_filter,omitempty"`
+	ReasonSearch                  string `json:"reason_search,omitempty"`
+	MessageSearch                 string `json:"message_search,omitempty"`
+	TypeFilter                    string `json:"type_filter,omitempty"`
+
+	StartTime *time.Time `json:"start_time,omitempty"`
+	EndTime   *time.Time `json:"end_time,omitempty"`
+
+	Interval string `json:"interval,omitempty"`
+	Sort     string `json:"sort,omitempty"`
+	From     int64  `json:"from,omitempty"`
+	Size     int64  `json:"size,omitempty"`
+}
+
+func ParseQueryParameter(req *restful.Request) (*Query, error) {
+	q := &Query{}
+
+	q.Operation = req.QueryParameter("operation")
+	q.WorkspaceFilter = req.QueryParameter("workspace_filter")
+	q.WorkspaceSearch = req.QueryParameter("workspace_search")
+	q.InvolvedObjectNamespaceFilter = req.QueryParameter("involved_object_namespace_filter")
+	q.InvolvedObjectNamespaceSearch = req.QueryParameter("involved_object_namespace_search")
+	q.InvolvedObjectNameFilter = req.QueryParameter("involved_object_name_filter")
+	q.InvolvedObjectNameSearch = req.QueryParameter("involved_object_name_search")
+	q.InvolvedObjectKindFilter = req.QueryParameter("involved_object_kind_filter")
+	q.ReasonFilter = req.QueryParameter("reason_filter")
+	q.ReasonSearch = req.QueryParameter("reason_search")
+	q.MessageSearch = req.QueryParameter("message_search")
+	q.TypeFilter = req.QueryParameter("type_filter")
+
+	if tstr := req.QueryParameter("start_time"); tstr != "" {
+		sec, err := strconv.ParseInt(tstr, 10, 64)
+		if err != nil {
+			return nil, err
+		}
+		t := time.Unix(sec, 0)
+		q.StartTime = &t
+	}
+	if tstr := req.QueryParameter("end_time"); tstr != "" {
+		sec, err := strconv.ParseInt(tstr, 10, 64)
+		if err != nil {
+			return nil, err
+		}
+		t := time.Unix(sec, 0)
+		q.EndTime = &t
+	}
+	if q.Interval = req.QueryParameter("interval"); q.Interval == "" {
+		q.Interval = "15m"
+	}
+	q.From, _ = strconv.ParseInt(req.QueryParameter("from"), 10, 64)
+	size, err := strconv.ParseInt(req.QueryParameter("size"), 10, 64)
+	if err != nil {
+		size = 10
+	}
+	q.Size = size
+	if q.Sort = req.QueryParameter("sort"); q.Sort != "asc" {
+		q.Sort = "desc"
+	}
+	return q, nil
+}

pkg/apiserver/apiserver.go

@@ -51,6 +51,7 @@ import (
 	"kubesphere.io/kubesphere/pkg/models/iam/im"
 	"kubesphere.io/kubesphere/pkg/simple/client/cache"
 	"kubesphere.io/kubesphere/pkg/simple/client/devops"
+	"kubesphere.io/kubesphere/pkg/simple/client/events"
 	"kubesphere.io/kubesphere/pkg/simple/client/k8s"
 	"kubesphere.io/kubesphere/pkg/simple/client/ldap"
 	"kubesphere.io/kubesphere/pkg/simple/client/logging"
@@ -118,6 +119,8 @@ type APIServer struct {
 	LdapClient ldap.Interface
 
 	SonarClient sonarqube.SonarInterface
+
+	EventsClient events.Client
 }
 
 func (s *APIServer) PrepareRun() error {
@@ -154,7 +157,7 @@ func (s *APIServer) installKubeSphereAPIs() {
 	urlruntime.Must(networkv1alpha2.AddToContainer(s.container, s.Config.NetworkOptions.WeaveScopeHost))
 	urlruntime.Must(operationsv1alpha2.AddToContainer(s.container, s.KubernetesClient.Kubernetes()))
 	urlruntime.Must(resourcesv1alpha2.AddToContainer(s.container, s.KubernetesClient.Kubernetes(), s.InformerFactory))
-	urlruntime.Must(tenantv1alpha2.AddToContainer(s.container, s.InformerFactory))
+	urlruntime.Must(tenantv1alpha2.AddToContainer(s.container, s.InformerFactory, s.EventsClient))
 	urlruntime.Must(terminalv1alpha2.AddToContainer(s.container, s.KubernetesClient.Kubernetes(), s.KubernetesClient.Config()))
 	urlruntime.Must(clusterkapisv1alpha1.AddToContainer(s.container,
 		s.InformerFactory.KubernetesSharedInformerFactory(),

pkg/apiserver/config/config.go

@@ -8,6 +8,7 @@ import (
 	"kubesphere.io/kubesphere/pkg/simple/client/alerting"
 	"kubesphere.io/kubesphere/pkg/simple/client/cache"
 	"kubesphere.io/kubesphere/pkg/simple/client/devops/jenkins"
+	eventsclient "kubesphere.io/kubesphere/pkg/simple/client/events/elasticsearch"
 	"kubesphere.io/kubesphere/pkg/simple/client/k8s"
 	"kubesphere.io/kubesphere/pkg/simple/client/ldap"
 	"kubesphere.io/kubesphere/pkg/simple/client/logging/elasticsearch"
@@ -74,6 +75,7 @@ type Config struct {
 	AuthenticationOptions *authoptions.AuthenticationOptions         `json:"authentication,omitempty" yaml:"authentication,omitempty" mapstructure:"authentication"`
 	AuthorizationOptions  *authorizationoptions.AuthorizationOptions `json:"authorization,omitempty" yaml:"authorization,omitempty" mapstructure:"authorization"`
 	MultiClusterOptions   *multicluster.Options                      `json:"multicluster,omitempty" yaml:"multicluster,omitempty" mapstructure:"multicluster"`
+	EventsOptions         *eventsclient.Options                      `json:"events,omitempty" yaml:"events,omitempty" mapstructure:"events"`
 	// Options used for enabling components, not actually used now. Once we switch Alerting/Notification API to kubesphere,
 	// we can add these options to kubesphere command lines
 	AlertingOptions     *alerting.Options     `json:"alerting,omitempty" yaml:"alerting,omitempty" mapstructure:"alerting"`
@@ -99,6 +101,7 @@ func New() *Config {
 		AuthenticationOptions: authoptions.NewAuthenticateOptions(),
 		AuthorizationOptions:  authorizationoptions.NewAuthorizationOptions(),
 		MultiClusterOptions:   multicluster.NewOptions(),
+		EventsOptions:         eventsclient.NewElasticSearchOptions(),
 	}
 }
 
@@ -213,4 +216,8 @@ func (conf *Config) stripEmptyOptions() {
 	if conf.MultiClusterOptions != nil && !conf.MultiClusterOptions.Enable {
 		conf.MultiClusterOptions = nil
 	}
+
+	if conf.EventsOptions != nil && conf.EventsOptions.Host == "" {
+		conf.EventsOptions = nil
+	}
 }

pkg/apiserver/config/config_test.go

@@ -11,6 +11,7 @@ import (
 	"kubesphere.io/kubesphere/pkg/simple/client/alerting"
 	"kubesphere.io/kubesphere/pkg/simple/client/cache"
 	"kubesphere.io/kubesphere/pkg/simple/client/devops/jenkins"
+	eventsclient "kubesphere.io/kubesphere/pkg/simple/client/events/elasticsearch"
 	"kubesphere.io/kubesphere/pkg/simple/client/k8s"
 	"kubesphere.io/kubesphere/pkg/simple/client/ldap"
 	"kubesphere.io/kubesphere/pkg/simple/client/logging/elasticsearch"
@@ -124,6 +125,11 @@ func newTestConfig() (*Config, error) {
 		MultiClusterOptions: &multicluster.Options{
 			Enable: false,
 		},
+		EventsOptions: &eventsclient.Options{
+			Host:        "http://elasticsearch-logging-data.kubesphere-logging-system.svc:9200",
+			IndexPrefix: "ks-logstash-events",
+			Version:     "6",
+		},
 	}
 	return conf, nil
 }

pkg/constants/constants.go

@@ -81,6 +81,7 @@ const (
 	CustomMetricsTag           = "Custom Metrics"
 	LogQueryTag                = "Log Query"
 	TerminalTag                = "Terminal"
+	EventsQueryTag             = "Events Query"
 )
 
 var (

pkg/kapis/tenant/v1alpha2/handler.go

@@ -5,20 +5,22 @@ import (
 	"github.com/emicklei/go-restful"
 	"k8s.io/klog"
 	"kubesphere.io/kubesphere/pkg/api"
+	eventsv1alpha1 "kubesphere.io/kubesphere/pkg/api/events/v1alpha1"
 	"kubesphere.io/kubesphere/pkg/apiserver/query"
 	"kubesphere.io/kubesphere/pkg/apiserver/request"
 	"kubesphere.io/kubesphere/pkg/informers"
 	"kubesphere.io/kubesphere/pkg/models/tenant"
+	"kubesphere.io/kubesphere/pkg/simple/client/events"
 )
 
 type tenantHandler struct {
 	tenant tenant.Interface
 }
 
-func newTenantHandler(factory informers.InformerFactory) *tenantHandler {
+func newTenantHandler(factory informers.InformerFactory, evtsClient events.Client) *tenantHandler {
 
 	return &tenantHandler{
-		tenant: tenant.New(factory),
+		tenant: tenant.New(factory, evtsClient),
 	}
 }
 
@@ -65,3 +67,29 @@ func (h *tenantHandler) ListNamespaces(req *restful.Request, resp *restful.Respo
 
 	resp.WriteEntity(result)
 }
+
+func (h *tenantHandler) Events(req *restful.Request, resp *restful.Response) {
+	user, ok := request.UserFrom(req.Request.Context())
+	if !ok {
+		err := errors.New("cannot obtain user info")
+		klog.Errorln(err)
+		api.HandleForbidden(resp, req, err)
+		return
+	}
+	queryParam, err := eventsv1alpha1.ParseQueryParameter(req)
+	if err != nil {
+		klog.Errorln(err)
+		api.HandleInternalError(resp, req, err)
+		return
+	}
+
+	result, err := h.tenant.Events(user, queryParam)
+	if err != nil {
+		klog.Errorln(err)
+		api.HandleInternalError(resp, req, err)
+		return
+	}
+
+	resp.WriteEntity(result)
+
+}

pkg/kapis/tenant/v1alpha2/register.go

@@ -23,10 +23,12 @@ import (
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"kubesphere.io/kubesphere/pkg/api"
+	eventsv1alpha1 "kubesphere.io/kubesphere/pkg/api/events/v1alpha1"
 	"kubesphere.io/kubesphere/pkg/apiserver/runtime"
 	"kubesphere.io/kubesphere/pkg/constants"
 	"kubesphere.io/kubesphere/pkg/informers"
 	"kubesphere.io/kubesphere/pkg/models"
+	"kubesphere.io/kubesphere/pkg/simple/client/events"
 	"net/http"
 )
 
@@ -36,9 +38,9 @@ const (
 
 var GroupVersion = schema.GroupVersion{Group: GroupName, Version: "v1alpha2"}
 
-func AddToContainer(c *restful.Container, factory informers.InformerFactory) error {
+func AddToContainer(c *restful.Container, factory informers.InformerFactory, evtsClient events.Client) error {
 	ws := runtime.NewWebService(GroupVersion)
-	handler := newTenantHandler(factory)
+	handler := newTenantHandler(factory, evtsClient)
 
 	ws.Route(ws.GET("/workspaces").
 		To(handler.ListWorkspaces).
@@ -52,6 +54,32 @@ func AddToContainer(c *restful.Container, factory informers.InformerFactory) err
 		Returns(http.StatusOK, api.StatusOK, []v1.Namespace{}).
 		Metadata(restfulspec.KeyOpenAPITags, []string{constants.TenantResourcesTag}))
 
+	ws.Route(ws.GET("/events").
+		To(handler.Events).
+		Doc("Query events against the cluster").
+		Param(ws.QueryParameter("operation", "Operation type. This can be one of four types: `query` (for querying events), `statistics` (for retrieving statistical data), `histogram` (for displaying events count by time interval). Defaults to query.").DefaultValue("query")).
+		Param(ws.QueryParameter("workspace_filter", "A comma-separated list of workspaces. This field restricts the query to specified workspaces. For example, the following filter matches the workspace my-ws and demo-ws: `my-ws,demo-ws`.")).
+		Param(ws.QueryParameter("workspace_search", "A comma-separated list of keywords. Differing from **workspace_filter**, this field performs fuzzy matching on workspaces. For example, the following value limits the query to workspaces whose name contains the word my(My,MY,...) *OR* demo(Demo,DemO,...): `my,demo`.")).
+		Param(ws.QueryParameter("involved_object_namespace_filter", "A comma-separated list of namespaces. This field restricts the query to specified `involvedObject.namespace`.")).
+		Param(ws.QueryParameter("involved_object_namespace_search", "A comma-separated list of keywords. Differing from **involved_object_namespace_filter**, this field performs fuzzy matching on `involvedObject.namespace`")).
+		Param(ws.QueryParameter("involved_object_name_filter", "A comma-separated list of names. This field restricts the query to specified `involvedObject.name`.")).
+		Param(ws.QueryParameter("involved_object_name_search", "A comma-separated list of keywords. Differing from **involved_object_name_filter**, this field performs fuzzy matching on `involvedObject.name`.")).
+		Param(ws.QueryParameter("involved_object_kind_filter", "A comma-separated list of kinds. This field restricts the query to specified `involvedObject.kind`.")).
+		Param(ws.QueryParameter("reason_filter", "A comma-separated list of reasons. This field restricts the query to specified `reason`.")).
+		Param(ws.QueryParameter("reason_search", "A comma-separated list of keywords. Differing from **reason_filter**, this field performs fuzzy matching on `reason`.")).
+		Param(ws.QueryParameter("message_search", "A comma-separated list of keywords. This field performs fuzzy matching on `message`.")).
+		Param(ws.QueryParameter("type_filter", "Type of event matching on `type`. This can be one of two types: `Warning`, `Normal`")).
+		Param(ws.QueryParameter("start_time", "Start time of query (limits `lastTimestamp`). The format is a string representing seconds since the epoch, eg. 1136214245.")).
+		Param(ws.QueryParameter("end_time", "End time of query (limits `lastTimestamp`). The format is a string representing seconds since the epoch, eg. 1136214245.")).
+		Param(ws.QueryParameter("interval", "Time interval. It requires **operation** is set to `histogram`. The format is [0-9]+[smhdwMqy]. Defaults to 15m (i.e. 15 min).").DefaultValue("15m")).
+		Param(ws.QueryParameter("sort", "Sort order. One of asc, desc. This field sorts events by `lastTimestamp`.").DataType("string").DefaultValue("desc")).
+		Param(ws.QueryParameter("from", "The offset from the result set. This field returns query results from the specified offset. It requires **operation** is set to `query`. Defaults to 0 (i.e. from the beginning of the result set).").DataType("integer").DefaultValue("0").Required(false)).
+		Param(ws.QueryParameter("size", "Size of result set to return. It requires **operation** is set to `query`. Defaults to 10 (i.e. 10 event records).").DataType("integer").DefaultValue("10").Required(false)).
+		Metadata(restfulspec.KeyOpenAPITags, []string{constants.EventsQueryTag}).
+		Writes(eventsv1alpha1.APIResponse{}).
+		Returns(http.StatusOK, api.StatusOK, eventsv1alpha1.APIResponse{}))
+
 	c.Add(ws)
+
 	return nil
 }

pkg/models/events/events.go

+package events
+
+import (
+	eventsv1alpha1 "kubesphere.io/kubesphere/pkg/api/events/v1alpha1"
+	"kubesphere.io/kubesphere/pkg/simple/client/events"
+	"kubesphere.io/kubesphere/pkg/utils/stringutils"
+)
+
+type Interface interface {
+	Events(queryParam *eventsv1alpha1.Query, MutateFilterFunc func(*events.Filter)) (*eventsv1alpha1.APIResponse, error)
+}
+
+type eventsOperator struct {
+	client events.Client
+}
+
+func NewEventsOperator(client events.Client) Interface {
+	return &eventsOperator{client}
+}
+
+func (eo *eventsOperator) Events(queryParam *eventsv1alpha1.Query,
+	MutateFilterFunc func(*events.Filter)) (*eventsv1alpha1.APIResponse, error) {
+	filter := &events.Filter{
+		InvolvedObjectNames:     stringutils.Split(queryParam.InvolvedObjectNameFilter, ","),
+		InvolvedObjectNameFuzzy: stringutils.Split(queryParam.InvolvedObjectNameSearch, ","),
+		InvolvedObjectkinds:     stringutils.Split(queryParam.InvolvedObjectKindFilter, ","),
+		Reasons:                 stringutils.Split(queryParam.ReasonFilter, ","),
+		ReasonFuzzy:             stringutils.Split(queryParam.ReasonSearch, ","),
+		MessageFuzzy:            stringutils.Split(queryParam.MessageSearch, ","),
+		Type:                    queryParam.TypeFilter,
+		StartTime:               queryParam.StartTime,
+		EndTime:                 queryParam.EndTime,
+	}
+	if MutateFilterFunc != nil {
+		MutateFilterFunc(filter)
+	}
+
+	var ar eventsv1alpha1.APIResponse
+	var err error
+	switch queryParam.Operation {
+	case "histogram":
+		if len(filter.InvolvedObjectNamespaceMap) == 0 {
+			ar.Histogram = &events.Histogram{}
+		} else {
+			ar.Histogram, err = eo.client.CountOverTime(filter, queryParam.Interval)
+		}
+	case "statistics":
+		if len(filter.InvolvedObjectNamespaceMap) == 0 {
+			ar.Statistics = &events.Statistics{}
+		} else {
+			ar.Statistics, err = eo.client.StatisticsOnResources(filter)
+		}
+	default:
+		if len(filter.InvolvedObjectNamespaceMap) == 0 {
+			ar.Events = &events.Events{}
+		} else {
+			ar.Events, err = eo.client.SearchEvents(filter, queryParam.From, queryParam.Size, queryParam.Sort)
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &ar, nil
+}

pkg/models/tenant/tenant.go

@@ -25,29 +25,37 @@ import (
 	"k8s.io/apiserver/pkg/authentication/user"
 	"k8s.io/klog"
 	"kubesphere.io/kubesphere/pkg/api"
+	eventsv1alpha1 "kubesphere.io/kubesphere/pkg/api/events/v1alpha1"
 	tenantv1alpha1 "kubesphere.io/kubesphere/pkg/apis/tenant/v1alpha1"
 	"kubesphere.io/kubesphere/pkg/apiserver/authorization/authorizer"
 	"kubesphere.io/kubesphere/pkg/apiserver/authorization/authorizerfactory"
 	unionauthorizer "kubesphere.io/kubesphere/pkg/apiserver/authorization/union"
 	"kubesphere.io/kubesphere/pkg/apiserver/query"
 	"kubesphere.io/kubesphere/pkg/informers"
+	"kubesphere.io/kubesphere/pkg/models/events"
 	"kubesphere.io/kubesphere/pkg/models/iam/am"
 	resources "kubesphere.io/kubesphere/pkg/models/resources/v1alpha3"
 	resourcesv1alpha3 "kubesphere.io/kubesphere/pkg/models/resources/v1alpha3/resource"
+	eventsclient "kubesphere.io/kubesphere/pkg/simple/client/events"
+	"kubesphere.io/kubesphere/pkg/utils/stringutils"
+	"strings"
+	"time"
 )
 
 type Interface interface {
 	ListWorkspaces(user user.Info, query *query.Query) (*api.ListResult, error)
 	ListNamespaces(user user.Info, workspace string, query *query.Query) (*api.ListResult, error)
+	Events(user user.Info, queryParam *eventsv1alpha1.Query) (*eventsv1alpha1.APIResponse, error)
 }
 
 type tenantOperator struct {
 	am             am.AccessManagementInterface
 	authorizer     authorizer.Authorizer
 	resourceGetter *resourcesv1alpha3.ResourceGetter
+	events         events.Interface
 }
 
-func New(informers informers.InformerFactory) Interface {
+func New(informers informers.InformerFactory, evtsClient eventsclient.Client) Interface {
 	amOperator := am.NewAMOperator(informers)
 	rbacAuthorizer := authorizerfactory.NewRBACAuthorizer(amOperator)
 	opaAuthorizer := authorizerfactory.NewOPAAuthorizer(amOperator)
@@ -56,6 +64,7 @@ func New(informers informers.InformerFactory) Interface {
 		am:             amOperator,
 		authorizer:     authorizers,
 		resourceGetter: resourcesv1alpha3.NewResourceGetter(informers),
+		events:         events.NewEventsOperator(evtsClient),
 	}
 }
 
@@ -200,6 +209,131 @@ func (t *tenantOperator) ListNamespaces(user user.Info, workspace string, queryP
 	return result, nil
 }
 
+// listIntersectedNamespaces lists the namespaces which meet all the following conditions at the same time
+// 1. the namespace which belongs to user.
+// 2. the namespace in workspace which is in workspaces when workspaces is not empty.
+// 3. the namespace in workspace which contains one of workspaceSubstrs when workspaceSubstrs is not empty.
+// 4. the namespace which is in namespaces when namespaces is not empty.
+// 5. the namespace which contains one of namespaceSubstrs when namespaceSubstrs is not empty.
+func (t *tenantOperator) listIntersectedNamespaces(user user.Info,
+	workspaces, workspaceSubstrs, namespaces, namespaceSubstrs []string) ([]*corev1.Namespace, error) {
+	var (
+		namespaceSet = stringSet(namespaces)
+		workspaceSet = stringSet(workspaces)
+
+		iNamespaces []*corev1.Namespace
+	)
+
+	// When user can list all namespaces, the namespaces which do not belong to any workspace should be considered
+	listNs := authorizer.AttributesRecord{
+		User:            user,
+		Verb:            "list",
+		APIGroup:        "",
+		APIVersion:      "v1",
+		Resource:        "namespaces",
+		ResourceRequest: true,
+	}
+	decision, _, err := t.authorizer.Authorize(listNs)
+	if err != nil {
+		return nil, err
+	}
+	includeNsWithoutWs := len(workspaceSet) == 0 && len(workspaceSubstrs) == 0 && decision == authorizer.DecisionAllow
+
+	roleBindings, err := t.am.ListRoleBindings(user.GetName(), "")
+	if err != nil {
+		return nil, err
+	}
+	for _, rb := range roleBindings {
+		if len(namespaceSet) > 0 {
+			if _, ok := namespaceSet[rb.Namespace]; !ok {
+				continue
+			}
+		}
+		if len(namespaceSubstrs) > 0 && !stringContains(rb.Namespace, namespaceSubstrs) {
+			continue
+		}
+		ns, err := t.resourceGetter.Get("namespaces", "", rb.Namespace)
+		if err != nil {
+			return nil, err
+		}
+		if ns, ok := ns.(*corev1.Namespace); ok {
+			if ws := ns.Labels[tenantv1alpha1.WorkspaceLabel]; ws != "" {
+				if len(workspaceSet) > 0 {
+					if _, ok := workspaceSet[ws]; !ok {
+						continue
+					}
+				}
+				if len(workspaceSubstrs) > 0 && !stringContains(ws, workspaceSubstrs) {
+					continue
+				}
+			} else if !includeNsWithoutWs {
+				continue
+			}
+			iNamespaces = append(iNamespaces, ns)
+		}
+	}
+	return iNamespaces, nil
+}
+
+func (t *tenantOperator) Events(user user.Info, queryParam *eventsv1alpha1.Query) (*eventsv1alpha1.APIResponse, error) {
+	iNamespaces, err := t.listIntersectedNamespaces(user,
+		stringutils.Split(queryParam.WorkspaceFilter, ","),
+		stringutils.Split(queryParam.WorkspaceSearch, ","),
+		stringutils.Split(queryParam.InvolvedObjectNamespaceFilter, ","),
+		stringutils.Split(queryParam.InvolvedObjectNamespaceSearch, ","))
+	if err != nil {
+		klog.Error(err)
+		return nil, err
+	}
+
+	namespaceCreateTimeMap := make(map[string]time.Time)
+
+	for _, ns := range iNamespaces {
+		listEvts := authorizer.AttributesRecord{
+			User:            user,
+			Verb:            "list",
+			APIGroup:        "",
+			APIVersion:      "v1",
+			Namespace:       ns.Name,
+			Resource:        "events",
+			ResourceRequest: true,
+		}
+		decision, _, err := t.authorizer.Authorize(listEvts)
+		if err != nil {
+			klog.Error(err)
+			return nil, err
+		}
+		if decision == authorizer.DecisionAllow {
+			namespaceCreateTimeMap[ns.Name] = ns.CreationTimestamp.Time
+		}
+	}
+	// If there are no ns and ws query conditions,
+	// those events with empty `involvedObject.namespace` will also be listed when user can list all events
+	if len(queryParam.WorkspaceFilter) == 0 && len(queryParam.InvolvedObjectNamespaceFilter) == 0 &&
+		len(queryParam.WorkspaceSearch) == 0 && len(queryParam.InvolvedObjectNamespaceSearch) == 0 {
+		listEvts := authorizer.AttributesRecord{
+			User:            user,
+			Verb:            "list",
+			APIGroup:        "",
+			APIVersion:      "v1",
+			Resource:        "events",
+			ResourceRequest: true,
+		}
+		decision, _, err := t.authorizer.Authorize(listEvts)
+		if err != nil {
+			klog.Error(err)
+			return nil, err
+		}
+		if decision == authorizer.DecisionAllow {
+			namespaceCreateTimeMap[""] = time.Time{}
+		}
+	}
+
+	return t.events.Events(queryParam, func(filter *eventsclient.Filter) {
+		filter.InvolvedObjectNamespaceMap = namespaceCreateTimeMap
+	})
+}
+
 func contains(objects []runtime.Object, object runtime.Object) bool {
 	for _, item := range objects {
 		if item == object {
@@ -208,3 +342,20 @@ func contains(objects []runtime.Object, object runtime.Object) bool {
 	}
 	return false
 }
+
+func stringSet(strs []string) map[string]struct{} {
+	m := make(map[string]struct{})
+	for _, str := range strs {
+		m[str] = struct{}{}
+	}
+	return m
+}
+
+func stringContains(str string, subStrs []string) bool {
+	for _, sub := range subStrs {
+		if strings.Contains(str, sub) {
+			return true
+		}
+	}
+	return false
+}

pkg/models/tenant/tenent_test.go

@@ -332,5 +332,5 @@ func prepare() Interface {
 			RoleBindings().Informer().GetIndexer().Add(roleBinding)
 	}
 
-	return New(fakeInformerFactory)
+	return New(fakeInformerFactory, nil)
 }

pkg/simple/client/events/elasticsearch/clients.go

+package elasticsearch
+
+import (
+	"fmt"
+	es5 "github.com/elastic/go-elasticsearch/v5"
+	es5api "github.com/elastic/go-elasticsearch/v5/esapi"
+	es6 "github.com/elastic/go-elasticsearch/v6"
+	es6api "github.com/elastic/go-elasticsearch/v6/esapi"
+	es7 "github.com/elastic/go-elasticsearch/v7"
+	es7api "github.com/elastic/go-elasticsearch/v7/esapi"
+	jsoniter "github.com/json-iterator/go"
+	"io"
+	"net/http"
+)
+
+type Request struct {
+	Index string
+	Body  io.Reader
+}
+
+type Response struct {
+	Hits         Hits                           `json:"hits"`
+	Aggregations map[string]jsoniter.RawMessage `json:"aggregations"`
+}
+
+type Hits struct {
+	Total int64               `json:"total"`
+	Hits  jsoniter.RawMessage `json:"hits"`
+}
+
+type Error struct {
+	Type   string `json:"type"`
+	Reason string `json:"reason"`
+	Status int    `json:"status"`
+}
+
+func (e Error) Error() string {
+	return fmt.Sprintf("%s %s: %s", http.StatusText(e.Status), e.Type, e.Reason)
+}
+
+type ClientV5 es5.Client
+
+func (c *ClientV5) ExSearch(r *Request) (*Response, error) {
+	return c.parse(c.Search(c.Search.WithIndex(r.Index), c.Search.WithBody(r.Body)))
+}
+func (c *ClientV5) parse(resp *es5api.Response, err error) (*Response, error) {
+	if err != nil {
+		return nil, fmt.Errorf("error getting response: %s", err)
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		return nil, fmt.Errorf(resp.String())
+	}
+	var r struct {
+		Hits struct {
+			Total int64               `json:"total"`
+			Hits  jsoniter.RawMessage `json:"hits"`
+		} `json:"hits"`
+		Aggregations map[string]jsoniter.RawMessage `json:"aggregations"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
+		return nil, fmt.Errorf("error parsing the response body: %s", err)
+	}
+	return &Response{
+		Hits:         Hits{Total: r.Hits.Total, Hits: r.Hits.Hits},
+		Aggregations: r.Aggregations,
+	}, nil
+}
+func (c *ClientV5) Version() (string, error) {
+	res, err := c.Info()
+	if err != nil {
+		return "", err
+	}
+	defer res.Body.Close()
+	if res.IsError() {
+		return "", fmt.Errorf(res.String())
+	}
+	var r map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
+		return "", fmt.Errorf("error parsing the response body: %s", err)
+	}
+	return fmt.Sprintf("%s", r["version"].(map[string]interface{})["number"]), nil
+}
+
+type ClientV6 es6.Client
+
+func (c *ClientV6) ExSearch(r *Request) (*Response, error) {
+	return c.parse(c.Search(c.Search.WithIndex(r.Index), c.Search.WithBody(r.Body)))
+}
+func (c *ClientV6) parse(resp *es6api.Response, err error) (*Response, error) {
+	if err != nil {
+		return nil, fmt.Errorf("error getting response: %s", err)
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		return nil, fmt.Errorf(resp.String())
+	}
+	var r struct {
+		Hits *struct {
+			Total int64               `json:"total"`
+			Hits  jsoniter.RawMessage `json:"hits"`
+		} `json:"hits"`
+		Aggregations map[string]jsoniter.RawMessage `json:"aggregations"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
+		return nil, fmt.Errorf("error parsing the response body: %s", err)
+	}
+	return &Response{
+		Hits:         Hits{Total: r.Hits.Total, Hits: r.Hits.Hits},
+		Aggregations: r.Aggregations,
+	}, nil
+}
+
+type ClientV7 es7.Client
+
+func (c *ClientV7) ExSearch(r *Request) (*Response, error) {
+	return c.parse(c.Search(c.Search.WithIndex(r.Index), c.Search.WithBody(r.Body)))
+}
+func (c *ClientV7) parse(resp *es7api.Response, err error) (*Response, error) {
+	if err != nil {
+		return nil, fmt.Errorf("error getting response: %s", err)
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		return nil, fmt.Errorf(resp.String())
+	}
+	var r struct {
+		Hits *struct {
+			Total struct {
+				Value int64 `json:"value"`
+			} `json:"total"`
+			Hits jsoniter.RawMessage `json:"hits"`
+		} `json:"hits"`
+		Aggregations map[string]jsoniter.RawMessage `json:"aggregations"`
+	}
+	if err := json.NewDecoder(resp.Body).Decode(&r); err != nil {
+		return nil, fmt.Errorf("error parsing the response body: %s", err)
+	}
+	return &Response{
+		Hits:         Hits{Total: r.Hits.Total.Value, Hits: r.Hits.Hits},
+		Aggregations: r.Aggregations,
+	}, nil
+}
+
+type client interface {
+	ExSearch(r *Request) (*Response, error)
+}

pkg/simple/client/events/elasticsearch/elasticsearch.go

+package elasticsearch
+
+import (
+	"bytes"
+	"fmt"
+	"strings"
+	"time"
+
+	es5 "github.com/elastic/go-elasticsearch/v5"
+	es6 "github.com/elastic/go-elasticsearch/v6"
+	es7 "github.com/elastic/go-elasticsearch/v7"
+	jsoniter "github.com/json-iterator/go"
+	corev1 "k8s.io/api/core/v1"
+	"kubesphere.io/kubesphere/pkg/simple/client/events"
+)
+
+var json = jsoniter.ConfigCompatibleWithStandardLibrary
+
+type Elasticsearch struct {
+	c    client
+	opts struct {
+		index string
+	}
+}
+
+func (es *Elasticsearch) SearchEvents(filter *events.Filter, from, size int64,
+	sort string) (*events.Events, error) {
+	queryPart := parseToQueryPart(filter)
+	if sort == "" {
+		sort = "desc"
+	}
+	sortPart := []map[string]interface{}{{
+		"lastTimestamp": map[string]string{"order": sort},
+	}}
+	b := map[string]interface{}{
+		"from":  from,
+		"size":  size,
+		"query": queryPart,
+		"sort":  sortPart,
+	}
+
+	body, err := json.Marshal(b)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := es.c.ExSearch(&Request{
+		Index: es.opts.index,
+		Body:  bytes.NewBuffer(body),
+	})
+	if err != nil || resp == nil {
+		return nil, err
+	}
+
+	var innerHits []struct {
+		*corev1.Event `json:"_source"`
+	}
+	if err := json.Unmarshal(resp.Hits.Hits, &innerHits); err != nil {
+		return nil, err
+	}
+	evts := events.Events{Total: resp.Hits.Total}
+	for _, hit := range innerHits {
+		evts.Records = append(evts.Records, hit.Event)
+	}
+	return &evts, nil
+}
+
+func (es *Elasticsearch) CountOverTime(filter *events.Filter, interval string) (*events.Histogram, error) {
+	if interval == "" {
+		interval = "15m"
+	}
+
+	queryPart := parseToQueryPart(filter)
+	aggName := "events_count_over_lasttimestamp"
+	aggsPart := map[string]interface{}{
+		aggName: map[string]interface{}{
+			"date_histogram": map[string]string{
+				"field":    "lastTimestamp",
+				"interval": interval,
+			},
+		},
+	}
+	b := map[string]interface{}{
+		"query": queryPart,
+		"aggs":  aggsPart,
+		"size":  0, // do not get docs
+	}
+
+	body, err := json.Marshal(b)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := es.c.ExSearch(&Request{
+		Index: es.opts.index,
+		Body:  bytes.NewBuffer(body),
+	})
+	if err != nil || resp == nil {
+		return nil, err
+	}
+
+	raw := resp.Aggregations[aggName]
+	var agg struct {
+		Buckets []struct {
+			KeyAsString string `json:"key_as_string"`
+			Key         int64  `json:"key"`
+			DocCount    int64  `json:"doc_count"`
+		} `json:"buckets"`
+	}
+	if err := json.Unmarshal(raw, &agg); err != nil {
+		return nil, err
+	}
+	histo := events.Histogram{Total: int64(len(agg.Buckets))}
+	for _, b := range agg.Buckets {
+		histo.Buckets = append(histo.Buckets,
+			events.Bucket{Time: b.Key, Count: b.DocCount})
+	}
+	return &histo, nil
+}
+
+func (es *Elasticsearch) StatisticsOnResources(filter *events.Filter) (*events.Statistics, error) {
+	queryPart := parseToQueryPart(filter)
+	aggName := "resources_count"
+	aggsPart := map[string]interface{}{
+		aggName: map[string]interface{}{
+			"cardinality": map[string]string{
+				"field": "involvedObject.uid.keyword",
+			},
+		},
+	}
+	b := map[string]interface{}{
+		"query": queryPart,
+		"aggs":  aggsPart,
+		"size":  0, // do not get docs
+	}
+
+	body, err := json.Marshal(b)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := es.c.ExSearch(&Request{
+		Index: es.opts.index,
+		Body:  bytes.NewBuffer(body),
+	})
+	if err != nil || resp == nil {
+		return nil, err
+	}
+
+	raw := resp.Aggregations[aggName]
+	var agg struct {
+		Value int64 `json:"value"`
+	}
+	if err := json.Unmarshal(raw, &agg); err != nil {
+		return nil, err
+	}
+
+	return &events.Statistics{
+		Resources: agg.Value,
+		Events:    resp.Hits.Total,
+	}, nil
+}
+
+func NewClient(options *Options) (*Elasticsearch, error) {
+	clientV5 := func() (*ClientV5, error) {
+		c, err := es5.NewClient(es5.Config{Addresses: []string{options.Host}})
+		if err != nil {
+			return nil, err
+		}
+		return (*ClientV5)(c), nil
+	}
+	clientV6 := func() (*ClientV6, error) {
+		c, err := es6.NewClient(es6.Config{Addresses: []string{options.Host}})
+		if err != nil {
+			return nil, err
+		}
+		return (*ClientV6)(c), nil
+	}
+	clientV7 := func() (*ClientV7, error) {
+		c, err := es7.NewClient(es7.Config{Addresses: []string{options.Host}})
+		if err != nil {
+			return nil, err
+		}
+		return (*ClientV7)(c), nil
+	}
+
+	var (
+		version = options.Version
+		es      = Elasticsearch{}
+		err     error
+	)
+	es.opts.index = fmt.Sprintf("%s*", options.IndexPrefix)
+
+	if options.Version == "" {
+		var c5 *ClientV5
+		if c5, err = clientV5(); err == nil {
+			if version, err = c5.Version(); err == nil {
+				es.c = c5
+			}
+		}
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	switch strings.Split(version, ".")[0] {
+	case "5":
+		if es.c == nil {
+			es.c, err = clientV5()
+		}
+	case "6":
+		es.c, err = clientV6()
+	case "7":
+		es.c, err = clientV7()
+	default:
+		err = fmt.Errorf("unsupported elasticsearch version %s", version)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &es, nil
+}
+
+func parseToQueryPart(f *events.Filter) interface{} {
+	if f == nil {
+		return nil
+	}
+	type BoolBody struct {
+		Filter             []map[string]interface{} `json:"filter,omitempty"`
+		Should             []map[string]interface{} `json:"should,omitempty"`
+		MinimumShouldMatch *int                     `json:"minimum_should_match,omitempty"`
+	}
+	var mini = 1
+	b := BoolBody{}
+	queryBody := map[string]interface{}{
+		"bool": &b,
+	}
+
+	if len(f.InvolvedObjectNamespaceMap) > 0 {
+		bi := BoolBody{MinimumShouldMatch: &mini}
+		for k, v := range f.InvolvedObjectNamespaceMap {
+			bi.Should = append(bi.Should, map[string]interface{}{
+				"bool": &BoolBody{
+					Filter: []map[string]interface{}{{
+						"match_phrase": map[string]string{"involvedObject.namespace.keyword": k},
+					}, {
+						"range": map[string]interface{}{
+							"lastTimestamp": map[string]interface{}{
+								"gte": v,
+							},
+						},
+					}},
+				},
+			})
+		}
+		if len(bi.Should) > 0 {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": &bi})
+		}
+	}
+
+	shouldBoolbody := func(mtype, fieldName string, fieldValues []string, fieldValueMutate func(string) string) *BoolBody {
+		bi := BoolBody{MinimumShouldMatch: &mini}
+		for _, v := range fieldValues {
+			if fieldValueMutate != nil {
+				v = fieldValueMutate(v)
+			}
+			bi.Should = append(bi.Should, map[string]interface{}{
+				mtype: map[string]string{fieldName: v},
+			})
+		}
+		if len(bi.Should) == 0 {
+			return nil
+		}
+		return &bi
+	}
+
+	if len(f.InvolvedObjectNames) > 0 {
+		if bi := shouldBoolbody("match_phrase", "involvedObject.name.keyword",
+			f.InvolvedObjectNames, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+	if len(f.InvolvedObjectNameFuzzy) > 0 {
+		if bi := shouldBoolbody("match_phrase_prefix", "involvedObject.name",
+			f.InvolvedObjectNameFuzzy, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+	if len(f.InvolvedObjectkinds) > 0 {
+		// involvedObject.kind is single word and here is not field keyword for case ignoring
+		if bi := shouldBoolbody("match_phrase", "involvedObject.kind",
+			f.InvolvedObjectkinds, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+	if len(f.Reasons) > 0 {
+		// reason is single word and here is not field keyword for case ignoring
+		if bi := shouldBoolbody("match_phrase", "reason",
+			f.Reasons, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+	if len(f.ReasonFuzzy) > 0 {
+		if bi := shouldBoolbody("wildcard", "reason",
+			f.ReasonFuzzy, func(s string) string {
+				return fmt.Sprintf("*" + s + "*")
+			}); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+	if len(f.MessageFuzzy) > 0 {
+		if bi := shouldBoolbody("match_phrase_prefix", "message",
+			f.MessageFuzzy, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+
+	if len(f.Type) > 0 {
+		// type is single word and here is not field keyword for case ignoring
+		if bi := shouldBoolbody("match_phrase", "type",
+			[]string{f.Type}, nil); bi != nil {
+			b.Filter = append(b.Filter, map[string]interface{}{"bool": bi})
+		}
+	}
+
+	if f.StartTime != nil || f.EndTime != nil {
+		m := make(map[string]*time.Time)
+		if f.StartTime != nil {
+			m["gte"] = f.StartTime
+		}
+		if f.EndTime != nil {
+			m["lte"] = f.EndTime
+		}
+		b.Filter = append(b.Filter, map[string]interface{}{
+			"range": map[string]interface{}{"lastTimestamp": m},
+		})
+
+	}
+
+	return queryBody
+}

pkg/simple/client/events/elasticsearch/elasticsearch_test.go

+package elasticsearch
+
+import (
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
+	"kubesphere.io/kubesphere/pkg/simple/client/events"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+func MockElasticsearchService(pattern string, fakeCode int, fakeResp string) *httptest.Server {
+	mux := http.NewServeMux()
+	mux.HandleFunc(pattern, func(res http.ResponseWriter, req *http.Request) {
+		res.WriteHeader(fakeCode)
+		res.Write([]byte(fakeResp))
+	})
+	return httptest.NewServer(mux)
+}
+
+func TestStatisticsOnResources(t *testing.T) {
+	var tests = []struct {
+		description   string
+		filter        events.Filter
+		fakeVersion   string
+		fakeCode      int
+		fakeResp      string
+		expected      events.Statistics
+		expectedError bool
+	}{{
+		description: "ES index exists",
+		filter:      events.Filter{},
+		fakeVersion: "6",
+		fakeCode:    200,
+		fakeResp: `
+{
+  "took": 16,
+  "timed_out": false,
+  "_shards": {
+    "total": 1,
+    "successful": 1,
+    "skipped": 0,
+    "failed": 0
+  },
+  "hits": {
+    "total": 10000,
+    "max_score": null,
+    "hits": [
+
+    ]
+  },
+  "aggregations": {
+    "resources_count": {
+      "value": 100
+    }
+  }
+}
+`,
+		expected: events.Statistics{
+			Events:    10000,
+			Resources: 100,
+		},
+		expectedError: false,
+	}, {
+		description: "ES index not exists",
+		filter:      events.Filter{},
+		fakeVersion: "6",
+		fakeCode:    404,
+		fakeResp: `
+{
+  "error": {
+    "root_cause": [
+      {
+        "type": "index_not_found_exception",
+        "reason": "no such index [events]",
+        "resource.type": "index_or_alias",
+        "resource.id": "events",
+        "index_uuid": "_na_",
+        "index": "events"
+      }
+    ],
+    "type": "index_not_found_exception",
+    "reason": "no such index [events]",
+    "resource.type": "index_or_alias",
+    "resource.id": "events",
+    "index_uuid": "_na_",
+    "index": "events"
+  },
+  "status": 404
+}
+`,
+		expectedError: true,
+	}}
+
+	for _, test := range tests {
+		t.Run(test.description, func(t *testing.T) {
+			mes := MockElasticsearchService("/", test.fakeCode, test.fakeResp)
+			defer mes.Close()
+
+			es, err := NewClient(&Options{Host: mes.URL, IndexPrefix: "ks-logstash-events", Version: "6"})
+
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			stats, err := es.StatisticsOnResources(&test.filter)
+
+			if test.expectedError {
+				if err == nil {
+					t.Fatalf("expected err like %s", test.fakeResp)
+				} else if !strings.Contains(err.Error(), strconv.Itoa(test.fakeCode)) {
+					t.Fatalf("err does not contain expected code: %d", test.fakeCode)
+				}
+			} else {
+				if err != nil {
+					t.Fatal(err)
+				} else if diff := cmp.Diff(stats, &test.expected); diff != "" {
+					t.Fatalf("%T differ (-got, +want): %s", test.expected, diff)
+				}
+			}
+		})
+	}
+}
+
+func TestParseToQueryPart(t *testing.T) {
+	q := `
+{
+	"bool": {
+		"filter": [
+			{
+				"bool": {
+					"should": [
+						{
+							"bool": {
+								"filter": [
+									{
+										"match_phrase": {
+											"involvedObject.namespace.keyword": "kubesphere-system"
+										}
+									},
+									{
+										"range": {
+											"lastTimestamp": {
+												"gte": "2020-01-01T01:01:01.000000001Z"
+											}
+										}
+									}
+								]
+							}
+						}
+					],
+					"minimum_should_match": 1
+				}
+			},
+			{
+				"bool": {
+					"should": [
+						{
+							"match_phrase_prefix": {
+								"involvedObject.name": "istio"
+							}
+						}
+					],
+					"minimum_should_match": 1
+				}
+			},
+			{
+				"bool": {
+					"should": [
+						{
+							"match_phrase": {
+								"reason": "unhealthy"
+							}
+						}
+					],
+					"minimum_should_match": 1
+				}
+			},
+			{
+				"range": {
+					"lastTimestamp": {
+						"gte": "2019-12-01T01:01:01.000000001Z"
+					}
+				}
+			}
+		]
+	}
+}
+`
+	nsCreateTime := time.Date(2020, time.Month(1), 1, 1, 1, 1, 1, time.UTC)
+	startTime := nsCreateTime.AddDate(0, -1, 0)
+
+	filter := &events.Filter{
+		InvolvedObjectNamespaceMap: map[string]time.Time{
+			"kubesphere-system": nsCreateTime,
+		},
+		InvolvedObjectNameFuzzy: []string{"istio"},
+		Reasons:                 []string{"unhealthy"},
+		StartTime:               &startTime,
+	}
+
+	qp := parseToQueryPart(filter)
+	bs, err := json.Marshal(qp)
+	if err != nil {
+		panic(err)
+	}
+
+	queryPart := &map[string]interface{}{}
+	if err := json.Unmarshal(bs, queryPart); err != nil {
+		panic(err)
+	}
+	expectedQueryPart := &map[string]interface{}{}
+	if err := json.Unmarshal([]byte(q), expectedQueryPart); err != nil {
+		panic(err)
+	}
+
+	assert.Equal(t, expectedQueryPart, queryPart)
+}

pkg/simple/client/events/elasticsearch/options.go

+package elasticsearch
+
+import (
+	"github.com/spf13/pflag"
+	"kubesphere.io/kubesphere/pkg/utils/reflectutils"
+)
+
+type Options struct {
+	Host        string `json:"host" yaml:"host"`
+	IndexPrefix string `json:"indexPrefix,omitempty" yaml:"indexPrefix"`
+	Version     string `json:"version" yaml:"version"`
+}
+
+func NewElasticSearchOptions() *Options {
+	return &Options{
+		Host:        "",
+		IndexPrefix: "ks-logstash-events",
+		Version:     "",
+	}
+}
+
+func (s *Options) ApplyTo(options *Options) {
+	if s.Host != "" {
+		reflectutils.Override(options, s)
+	}
+}
+
+func (s *Options) Validate() []error {
+	errs := []error{}
+
+	return errs
+}
+
+func (s *Options) AddFlags(fs *pflag.FlagSet, c *Options) {
+	fs.StringVar(&s.Host, "elasticsearch-host", c.Host, ""+
+		"Elasticsearch service host. KubeSphere is using elastic as event store, "+
+		"if this filed left blank, KubeSphere will use kubernetes builtin event API instead, and"+
+		" the following elastic search options will be ignored.")
+
+	fs.StringVar(&s.IndexPrefix, "index-prefix", c.IndexPrefix, ""+
+		"Index name prefix. KubeSphere will retrieve events against indices matching the prefix.")
+
+	fs.StringVar(&s.Version, "elasticsearch-version", c.Version, ""+
+		"Elasticsearch major version, e.g. 5/6/7, if left blank, will detect automatically."+
+		"Currently, minimum supported version is 5.x")
+}

pkg/simple/client/events/interface.go

+package events
+
+import (
+	v1 "k8s.io/api/core/v1"
+	"time"
+)
+
+type Client interface {
+	SearchEvents(filter *Filter, from, size int64, sort string) (*Events, error)
+	CountOverTime(filter *Filter, interval string) (*Histogram, error)
+	StatisticsOnResources(filter *Filter) (*Statistics, error)
+}
+
+type Filter struct {
+	InvolvedObjectNamespaceMap map[string]time.Time
+	InvolvedObjectNames        []string
+	InvolvedObjectNameFuzzy    []string
+	InvolvedObjectkinds        []string
+	Reasons                    []string
+	ReasonFuzzy                []string
+	MessageFuzzy               []string
+	Type                       string
+	StartTime                  *time.Time
+	EndTime                    *time.Time
+}
+
+type Events struct {
+	Total   int64       `json:"total" description:"total number of matched results"`
+	Records []*v1.Event `json:"records" description:"actual array of results"`
+}
+
+type Histogram struct {
+	Total   int64    `json:"total" description:"total number of events"`
+	Buckets []Bucket `json:"buckets" description:"actual array of histogram results"`
+}
+type Bucket struct {
+	Time  int64 `json:"time" description:"timestamp"`
+	Count int64 `json:"count" description:"total number of events at intervals"`
+}
+
+type Statistics struct {
+	Resources int64 `json:"resources" description:"total number of resources"`
+	Events    int64 `json:"events" description:"total number of events"`
+}

tools/cmd/doc-gen/main.go

@@ -113,14 +113,14 @@ func generateSwaggerJson() []byte {
 	informerFactory := informers.NewNullInformerFactory()
 
 	urlruntime.Must(devopsv1alpha2.AddToContainer(container, informerFactory.KubeSphereSharedInformerFactory(), &fake.Devops{}, nil, clientsets.KubeSphere(), fakes3.NewFakeS3()))
-	urlruntime.Must(iamv1alpha2.AddToContainer(container, im.NewOperator(clientsets.KubeSphere(), informerFactory.KubeSphereSharedInformerFactory()), am.NewAMOperator(clientsets.KubeSphere(), informerFactory.KubeSphereSharedInformerFactory()), authoptions.NewAuthenticateOptions()))
+	urlruntime.Must(iamv1alpha2.AddToContainer(container, im.NewOperator(clientsets.KubeSphere(), informerFactory), am.NewAMOperator(informerFactory), authoptions.NewAuthenticateOptions()))
 	urlruntime.Must(loggingv1alpha2.AddToContainer(container, clientsets, nil))
 	urlruntime.Must(monitoringv1alpha3.AddToContainer(container, clientsets.Kubernetes(), nil))
 	urlruntime.Must(openpitrixv1.AddToContainer(container, informerFactory, nil))
 	urlruntime.Must(operationsv1alpha2.AddToContainer(container, clientsets.Kubernetes()))
 	urlruntime.Must(resourcesv1alpha2.AddToContainer(container, clientsets.Kubernetes(), informerFactory))
 	urlruntime.Must(resourcesv1alpha3.AddToContainer(container, informerFactory))
-	urlruntime.Must(tenantv1alpha2.AddToContainer(container, clientsets, informerFactory))
+	urlruntime.Must(tenantv1alpha2.AddToContainer(container, informerFactory, nil))
 	urlruntime.Must(terminalv1alpha2.AddToContainer(container, clientsets.Kubernetes(), nil))
 	urlruntime.Must(metricsv1alpha2.AddToContainer(container))
 	urlruntime.Must(networkv1alpha2.AddToContainer(container, ""))