server

Documentation

Index

• Constants
• Variables
• func InitConfManagers()
• func Read() (osrelease map[string]string, err error)
• func ReadFile(filename string) (osrelease map[string]string, err error)
• func ReadString(content string) (osrelease map[string]string, err error)
• func StartUp(Version string)
• type ConfManager
• type Configuration
• type ConnectionBody
• type ImportConfigurationStatus
• type InstallScriptGetter
• type Metrics
  • func NewMetrics() *Metrics
  • func (m *Metrics) GetCpuUsage() string
  • func (m *Metrics) GetMemoryUsage() string
• type RuleMigrationProcessor
  • func NewRuleMigrationProcessor(r *processor.RuleProcessor, s *processor.StreamProcessor) *RuleMigrationProcessor
  • func (p *RuleMigrationProcessor) ConfigurationPartialExport(rules []string) ([]byte, error)
• type RuleRegistry
  • func (rr *RuleRegistry) Delete(key string) (*rule.RuleState, bool)
  • func (rr *RuleRegistry) Load(key string) (value *rule.RuleState, ok bool)
  • func (rr *RuleRegistry) Store(key string, value *rule.RuleState)
• type RuleStatusMetricsValue
• type Server
  • func (t *Server) CreatePlugin(arg *model.PluginDesc, reply *string) error
  • func (t *Server) CreateQuery(sql string, reply *string) error
  • func (t *Server) CreateRule(rule *model.RPCArgDesc, reply *string) error
  • func (t *Server) CreateSchema(arg *model.RPCTypedArgDesc, reply *string) error
  • func (t *Server) CreateService(arg *model.RPCArgDesc, reply *string) error
  • func (t *Server) DescPlugin(arg *model.PluginDesc, reply *string) error
  • func (t *Server) DescRule(name string, reply *string) error
  • func (t *Server) DescSchema(arg *model.RPCTypedArgDesc, reply *string) error
  • func (t *Server) DescService(name string, reply *string) error
  • func (t *Server) DescServiceFunc(name string, reply *string) error
  • func (t *Server) DescUdf(arg string, reply *string) error
  • func (t *Server) DropPlugin(arg *model.PluginDesc, reply *string) error
  • func (t *Server) DropRule(name string, reply *string) error
  • func (t *Server) DropSchema(arg *model.RPCTypedArgDesc, reply *string) error
  • func (t *Server) DropService(name string, reply *string) error
  • func (t *Server) Export(file string, reply *string) error
  • func (t *Server) ExportConfiguration(arg *model.ExportDataDesc, reply *string) error
  • func (t *Server) GetQueryResult(_ string, reply *string) error
  • func (t *Server) GetStatusImport(_ int, reply *string) error
  • func (t *Server) GetStatusRule(name string, reply *string) error
  • func (t *Server) GetTopoRule(name string, reply *string) error
  • func (t *Server) Import(file string, reply *string) error
  • func (t *Server) ImportConfiguration(arg *model.ImportDataDesc, reply *string) error
  • func (t *Server) RegisterPlugin(arg *model.PluginDesc, reply *string) error
  • func (t *Server) RestartRule(name string, reply *string) error
  • func (t *Server) ShowPlugins(arg int, reply *string) error
  • func (t *Server) ShowRules(_ int, reply *string) error
  • func (t *Server) ShowSchemas(schemaType string, reply *string) error
  • func (t *Server) ShowServiceFuncs(_ int, reply *string) error
  • func (t *Server) ShowServices(_ int, reply *string) error
  • func (t *Server) ShowUdfs(_ int, reply *string) error
  • func (t *Server) StartRule(name string, reply *string) error
  • func (t *Server) StopRule(name string, reply *string) error
  • func (t *Server) Stream(stream string, reply *string) error
  • func (t *Server) ValidateRule(rule *model.RPCArgDesc, reply *string) error
• type UpdateRuleStateType

Constants

const (
	EtcOsRelease    string = "/etc/os-release"
	UsrLibOsRelease string = "/usr/lib/os-release"
)

const (
	ContentType     = "Content-Type"
	ContentTypeJSON = "application/json"
)

const QueryRuleId = "internal-ekuiper_query_rule"

Variables

var (
	NativeSourcePlugin   = []string{"random", "zmq", "sql", "video", "kafka"}
	NativeSinkPlugin     = []string{"image", "influx", "influx2", "tdengine", "zmq", "kafka", "sql"}
	NativeFunctionPlugin = []string{"accumulateWordCount", "countPlusOne", "echo", "geohash", "image", "labelImage", "tfLite"}
)

Functions

func InitConfManagers

func InitConfManagers()

func Read

func Read() (osrelease map[string]string, err error)

Read and return os-release, trying EtcOsRelease, followed by UsrLibOsRelease. err will contain an error message if neither file exists or failed to parse

func ReadFile

func ReadFile(filename string) (osrelease map[string]string, err error)

Similar to Read(), but takes the name of a file to load instead

func ReadString

func ReadString(content string) (osrelease map[string]string, err error)

ReadString is similar to Read(), but takes a string to load instead

func StartUp

func StartUp(Version string)

Types

type ConfManager

type ConfManager interface {
	Import(context.Context, map[string]string) map[string]string
	PartialImport(context.Context, map[string]string) map[string]string
	Export() map[string]string
	Status() map[string]string
	Reset()
}

type Configuration

type Configuration struct {
	Streams          map[string]string `json:"streams"`
	Tables           map[string]string `json:"tables"`
	Rules            map[string]string `json:"rules"`
	NativePlugins    map[string]string `json:"nativePlugins"`
	PortablePlugins  map[string]string `json:"portablePlugins"`
	SourceConfig     map[string]string `json:"sourceConfig"`
	SinkConfig       map[string]string `json:"sinkConfig"`
	ConnectionConfig map[string]string `json:"connectionConfig"`
	Service          map[string]string `json:"Service"`
	Schema           map[string]string `json:"Schema"`
	Uploads          map[string]string `json:"uploads"`
	Scripts          map[string]string `json:"scripts"`
}

type ConnectionBody

type ConnectionBody struct {
	Endpoint string `json:"endpoint"`
}

type ImportConfigurationStatus

type ImportConfigurationStatus struct {
	ErrorMsg       string
	ConfigResponse Configuration
}

type InstallScriptGetter

type InstallScriptGetter interface {
	InstallScript(s string) (string, string)
}

type Metrics

type Metrics struct {
	// contains filtered or unexported fields
}

func NewMetrics

func NewMetrics() *Metrics

func (*Metrics) GetCpuUsage

func (m *Metrics) GetCpuUsage() string

func (*Metrics) GetMemoryUsage

func (m *Metrics) GetMemoryUsage() string

type RuleMigrationProcessor

type RuleMigrationProcessor struct {
	// contains filtered or unexported fields
}

func NewRuleMigrationProcessor

func NewRuleMigrationProcessor(r *processor.RuleProcessor, s *processor.StreamProcessor) *RuleMigrationProcessor

func (*RuleMigrationProcessor) ConfigurationPartialExport

func (p *RuleMigrationProcessor) ConfigurationPartialExport(rules []string) ([]byte, error)

type RuleRegistry

type RuleRegistry struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*RuleRegistry) Delete

func (rr *RuleRegistry) Delete(key string) (*rule.RuleState, bool)

Delete Atomic get and delete. Only run when deleting a rule in runtime.

func (*RuleRegistry) Load

func (rr *RuleRegistry) Load(key string) (value *rule.RuleState, ok bool)

Load the entry of a rule by id. It is used to get the current rule state or send command to a running rule

func (*RuleRegistry) Store

func (rr *RuleRegistry) Store(key string, value *rule.RuleState)

Store create the in memory entry for a rule. Run in: 1. Restore the rules from KV at startup 2. Restore the rules when importing 3. Create a rule

type RuleStatusMetricsValue

type RuleStatusMetricsValue int

const (
	RuleStoppedByError RuleStatusMetricsValue = -1
	RuleStopped        RuleStatusMetricsValue = 0
	RuleRunning        RuleStatusMetricsValue = 1
)

type Server

type Server int

func (*Server) CreatePlugin

func (t *Server) CreatePlugin(arg *model.PluginDesc, reply *string) error

func (*Server) CreateQuery

func (t *Server) CreateQuery(sql string, reply *string) error

func (*Server) CreateRule

func (t *Server) CreateRule(rule *model.RPCArgDesc, reply *string) error

func (*Server) CreateSchema

func (t *Server) CreateSchema(arg *model.RPCTypedArgDesc, reply *string) error

func (*Server) CreateService

func (t *Server) CreateService(arg *model.RPCArgDesc, reply *string) error

func (*Server) DescPlugin

func (t *Server) DescPlugin(arg *model.PluginDesc, reply *string) error

func (*Server) DescRule

func (t *Server) DescRule(name string, reply *string) error

func (*Server) DescSchema

func (t *Server) DescSchema(arg *model.RPCTypedArgDesc, reply *string) error

func (*Server) DescService

func (t *Server) DescService(name string, reply *string) error

func (*Server) DescServiceFunc

func (t *Server) DescServiceFunc(name string, reply *string) error

func (*Server) DescUdf

func (t *Server) DescUdf(arg string, reply *string) error

func (*Server) DropPlugin

func (t *Server) DropPlugin(arg *model.PluginDesc, reply *string) error

func (*Server) DropRule

func (t *Server) DropRule(name string, reply *string) error

func (*Server) DropSchema

func (t *Server) DropSchema(arg *model.RPCTypedArgDesc, reply *string) error

func (*Server) DropService

func (t *Server) DropService(name string, reply *string) error

func (*Server) Export

func (t *Server) Export(file string, reply *string) error

func (*Server) ExportConfiguration

func (t *Server) ExportConfiguration(arg *model.ExportDataDesc, reply *string) error

func (*Server) GetQueryResult

func (t *Server) GetQueryResult(_ string, reply *string) error

*

• qid is not currently used.

func (*Server) GetStatusImport

func (t *Server) GetStatusImport(_ int, reply *string) error

func (*Server) GetStatusRule

func (t *Server) GetStatusRule(name string, reply *string) error

func (*Server) GetTopoRule

func (t *Server) GetTopoRule(name string, reply *string) error

func (*Server) Import

func (t *Server) Import(file string, reply *string) error

func (*Server) ImportConfiguration

func (t *Server) ImportConfiguration(arg *model.ImportDataDesc, reply *string) error

func (*Server) RegisterPlugin

func (t *Server) RegisterPlugin(arg *model.PluginDesc, reply *string) error

func (*Server) RestartRule

func (t *Server) RestartRule(name string, reply *string) error

func (*Server) ShowPlugins

func (t *Server) ShowPlugins(arg int, reply *string) error

func (*Server) ShowRules

func (t *Server) ShowRules(_ int, reply *string) error

func (*Server) ShowSchemas

func (t *Server) ShowSchemas(schemaType string, reply *string) error

func (*Server) ShowServiceFuncs

func (t *Server) ShowServiceFuncs(_ int, reply *string) error

func (*Server) ShowServices

func (t *Server) ShowServices(_ int, reply *string) error

func (*Server) ShowUdfs

func (t *Server) ShowUdfs(_ int, reply *string) error

func (*Server) StartRule

func (t *Server) StartRule(name string, reply *string) error

func (*Server) StopRule

func (t *Server) StopRule(name string, reply *string) error

func (*Server) Stream

func (t *Server) Stream(stream string, reply *string) error

func (*Server) ValidateRule

func (t *Server) ValidateRule(rule *model.RPCArgDesc, reply *string) error

type UpdateRuleStateType

type UpdateRuleStateType int

const (
	UpdateRuleState UpdateRuleStateType = iota
	UpdateRuleOffset
)