feat(region,host): host dmesg collect (#23747)

cmd/climc/shell/events/events.go

@@ -83,6 +83,10 @@ func doK8sEventList(s *mcclient.ClientSession, args *EventListOptions) error {
 	return DoEventList(*k8s.Logs.ResourceManager, s, args)
 }
 
+func doHostDmesgList(s *mcclient.ClientSession, args *EventListOptions) error {
+	return DoEventList(modules.DmesgLogs, s, args)
+}
+
 func DoEventList(man modulebase.ResourceManager, s *mcclient.ClientSession, args *EventListOptions) error {
 	params := jsonutils.NewDict()
 	if len(args.Type) > 0 {
@@ -157,6 +161,7 @@ func init() {
 	R(&EventListOptions{}, "monitor-event-show", "Show operation event logs", doMonitorEventList)
 	R(&EventListOptions{}, "notify-event-show", "Show operation event logs", doNotifyEventList)
 	R(&EventListOptions{}, "kube-event-show", "Show operation event logs", doK8sEventList)
+	R(&EventListOptions{}, "host-dmesg-show", "Show host dmesgs", doHostDmesgList)
 
 	R(&TypeEventListOptions{}, "server-event", "Show operation event logs of server", func(s *mcclient.ClientSession, args *TypeEventListOptions) error {
 		nargs := EventListOptions{BaseEventListOptions: args.BaseEventListOptions, Id: args.ID, Type: []string{"server"}}

pkg/apis/compute/host.go

@@ -629,6 +629,17 @@ type SHostPingInput struct {
 	QgaRunningGuestIds []string `json:"qga_running_guests"`
 }
 
+type SKmsgEntry struct {
+	Level   int       `json:"level"`
+	Seq     int       `json:"sql"`
+	Message string    `json:"message"`
+	Time    time.Time `json:"time"`
+}
+
+type SHostReportDmesgInput struct {
+	Entries []SKmsgEntry `json:"entries"`
+}
+
 type HostReserveCpusInput struct {
 	Cpus                    string
 	Mems                    string

pkg/apis/list.go

@@ -383,6 +383,14 @@ type OpsLogListInput struct {
 	Until time.Time `json:"until"`
 }
 
+type HostDmesgLogListInput struct {
+	OpsLogListInput
+
+	LogLevels []string `json:"log_levels"`
+
+	ShowDmesgLog bool `json:"show_dmesg_log"`
+}
+
 type IdNameDetails struct {
 	Id   string `json:"id"`
 	Name string `json:"name"`

pkg/cloudcommon/db/opslog.go

@@ -332,6 +332,7 @@ func (manager *SOpsLogManager) ListItemFilter(
 			q = q.Filter(sqlchemy.In(q.Field("action"), input.Actions))
 		}
 	}
+
 	//if !IsAdminAllowList(userCred, manager) {
 	// 	q = q.Filter(sqlchemy.OR(
 	//		sqlchemy.Equals(q.Field("owner_tenant_id"), manager.GetOwnerId(userCred)),

pkg/cloudcommon/db/opslog_const.go

@@ -346,3 +346,40 @@ const (
 
 	ACT_SET_COMMIT_BOUND = "set_commit_bound"
 )
+
+const (
+	ACT_HOST_DMESG = "host_dmesg"
+
+	LOGLEVEL_EMERG   = "emerg"
+	LOGLEVEL_ALERT   = "alert"
+	LOGLEVEL_CRIT    = "crit"
+	LOGLEVEL_ERR     = "err"
+	LOGLEVEL_WARNING = "warning"
+	LOGLEVEL_NOTICE  = "notice"
+	LOGLEVEL_INFO    = "info"
+	LOGLEVEL_DEBUG   = "debug"
+	LOGLEVEL_UNKNOWN = "unknown"
+)
+
+func LogLevelToString(logLevel int) string {
+	switch logLevel {
+	case 0:
+		return LOGLEVEL_EMERG
+	case 1:
+		return LOGLEVEL_ALERT
+	case 2:
+		return LOGLEVEL_CRIT
+	case 3:
+		return LOGLEVEL_ERR
+	case 4:
+		return LOGLEVEL_WARNING
+	case 5:
+		return LOGLEVEL_NOTICE
+	case 6:
+		return LOGLEVEL_INFO
+	case 7:
+		return LOGLEVEL_DEBUG
+	default:
+		return LOGLEVEL_UNKNOWN
+	}
+}

pkg/compute/models/host_dmesg.go

@@ -0,0 +1,106 @@
+// Copyright 2019 Yunion
+//
+// 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 models
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/util/stringutils"
+	"yunion.io/x/sqlchemy"
+
+	"yunion.io/x/onecloud/pkg/apis"
+	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/mcclient"
+)
+
+type SHostDmesgLogManager struct {
+	db.SOpsLogManager
+}
+
+type SHostDmesgLog struct {
+	db.SOpsLog
+
+	LogLevel string `width:"8" charset:"ascii" nullable:"true" list:"user" create:"optional"`
+}
+
+var HostDmesgLogManager *SHostDmesgLogManager
+
+var _ db.IModelManager = (*SHostDmesgLogManager)(nil)
+var _ db.IModel = (*SHostDmesgLog)(nil)
+
+func NewHostDmesgLogManager(opslog interface{}, tblName string, keyword, keywordPlural string, timeField string, clickhouse bool) SHostDmesgLogManager {
+	return SHostDmesgLogManager{
+		SOpsLogManager: db.NewOpsLogManager(opslog, tblName, keyword, keywordPlural, timeField, clickhouse),
+	}
+}
+
+func init() {
+	tmp := NewHostDmesgLogManager(SHostDmesgLog{}, "hostdmesg_log_tbl", "hostdmesg", "hostdmesgs", "ops_time", consts.OpsLogWithClickhouse)
+	HostDmesgLogManager = &tmp
+	HostDmesgLogManager.SetVirtualObject(HostDmesgLogManager)
+}
+
+func (manager *SHostDmesgLogManager) LogDmesg(ctx context.Context, host *SHost, logLevel string, opsTime time.Time, notes interface{}, userCred mcclient.TokenCredential) {
+	dmesgLog := &SHostDmesgLog{}
+	dmesgLog.OpsTime = opsTime
+	dmesgLog.LogLevel = logLevel
+	dmesgLog.ObjId = host.GetId()
+	dmesgLog.ObjName = host.GetName()
+	dmesgLog.Action = db.ACT_HOST_DMESG
+	dmesgLog.ObjType = host.Keyword()
+	dmesgLog.Notes = stringutils.Interface2String(notes)
+	dmesgLog.ProjectId = userCred.GetProjectId()
+	dmesgLog.Project = userCred.GetProjectName()
+	dmesgLog.ProjectDomainId = userCred.GetProjectDomainId()
+	dmesgLog.ProjectDomain = userCred.GetProjectDomain()
+	dmesgLog.UserId = userCred.GetUserId()
+	dmesgLog.User = userCred.GetUserName()
+	dmesgLog.DomainId = userCred.GetDomainId()
+	dmesgLog.Domain = userCred.GetDomainName()
+	dmesgLog.Roles = strings.Join(userCred.GetRoles(), ",")
+
+	dmesgLog.SetModelManager(manager, dmesgLog)
+	err := manager.TableSpec().Insert(ctx, dmesgLog)
+	if err != nil {
+		log.Errorf("fail to insert dmesgLog %s", err)
+	}
+}
+
+func (manager *SHostDmesgLogManager) ListItemFilter(
+	ctx context.Context,
+	q *sqlchemy.SQuery,
+	userCred mcclient.TokenCredential,
+	input apis.HostDmesgLogListInput,
+) (*sqlchemy.SQuery, error) {
+	var err error
+	q, err = manager.SOpsLogManager.ListItemFilter(ctx, q, userCred, input.OpsLogListInput)
+	if err != nil {
+		return q, err
+	}
+
+	if len(input.LogLevels) > 0 {
+		if len(input.LogLevels) == 1 {
+			q = q.Filter(sqlchemy.Equals(q.Field("log_level"), input.LogLevels[0]))
+		} else {
+			q = q.Filter(sqlchemy.In(q.Field("log_level"), input.LogLevels))
+		}
+	}
+
+	return q, nil
+}

pkg/compute/models/hosts.go

@@ -5229,6 +5229,14 @@ func (hh *SHost) GetStoragesByMasterHost() ([]string, error) {
 	return storages, nil
 }
 
+func (hh *SHost) PerformReportDmesg(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.SHostReportDmesgInput) (jsonutils.JSONObject, error) {
+	for i := range input.Entries {
+		logLevel := db.LogLevelToString(input.Entries[i].Level)
+		HostDmesgLogManager.LogDmesg(ctx, hh, logLevel, input.Entries[i].Time, input.Entries[i].Message, userCred)
+	}
+	return nil, nil
+}
+
 func (hh *SHost) PerformPing(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.SHostPingInput) (jsonutils.JSONObject, error) {
 	if hh.HostType == api.HOST_TYPE_BAREMETAL {
 		return nil, httperrors.NewNotSupportedError("ping host type %s not support", hh.HostType)

pkg/compute/models/initdb.go

@@ -52,6 +52,7 @@ func InitDB() error {
 		GuestManager,
 		GuestnetworkManager,
 		HostManager,
+		HostDmesgLogManager,
 		LoadbalancerCertificateManager,
 		LoadbalancerAclManager,
 		LoadbalancerManager,

pkg/compute/service/handlers.go

@@ -129,6 +129,7 @@ func InitHandlers(app *appsrv.Application) {
 		models.StoragecacheManager,
 		models.CachedimageManager,
 		models.HostManager,
+		models.HostDmesgLogManager,
 		models.SchedtagManager,
 		models.GuestManager,
 		models.GetContainerManager(),

pkg/hostman/hostinfo/hostinfo.go

@@ -2688,6 +2688,14 @@ func (h *SHostInfo) getNicsTelegrafConf() []map[string]interface{} {
 	return ret
 }
 
+func (h *SHostInfo) ReportHostDmesg(entries []api.SKmsgEntry) error {
+	data := api.SHostReportDmesgInput{
+		Entries: entries,
+	}
+	_, err := modules.Hosts.PerformAction(h.GetSession(), h.HostId, "report-dmesg", jsonutils.Marshal(data))
+	return err
+}
+
 func (h *SHostInfo) getHostname() string {
 	if h.FullName == "" {
 		h.FullName = h.fetchHostname()

pkg/hostman/hostmetrics/host_dmesg.go

@@ -0,0 +1,238 @@
+package hostmetrics
+
+import (
+	"bufio"
+	"fmt"
+	"os"
+	"path"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	"yunion.io/x/onecloud/pkg/apis/compute"
+	"yunion.io/x/onecloud/pkg/hostman/options"
+	"yunion.io/x/onecloud/pkg/util/fileutils2"
+)
+
+const (
+	kmsgPath      = "/dev/kmsg"
+	batchSize     = 100
+	flushInterval = 100 * time.Second
+)
+
+type SHostDmesgCollector struct {
+	host IHostInfo
+
+	mu       sync.Mutex
+	buffer   []compute.SKmsgEntry
+	bootTime time.Time
+}
+
+func NewHostDmesgCollector(hostInfo IHostInfo) *SHostDmesgCollector {
+	return &SHostDmesgCollector{
+		host:   hostInfo,
+		mu:     sync.Mutex{},
+		buffer: make([]compute.SKmsgEntry, 0),
+	}
+}
+
+// /dev/kmsg
+// level;sequence,timestamp[us];message
+// include/linux/kern_levels.h
+// #define LOGLEVEL_EMERG		0	/* system is unusable */
+// #define LOGLEVEL_ALERT		1	/* action must be taken immediately */
+// #define LOGLEVEL_CRIT		2	/* critical conditions */
+// #define LOGLEVEL_ERR		3	/* error conditions */
+// #define LOGLEVEL_WARNING	4	/* warning conditions */
+// #define LOGLEVEL_NOTICE		5	/* normal but significant condition */
+// #define LOGLEVEL_INFO		6	/* informational */
+// #define LOGLEVEL_DEBUG		7	/* debug-level messages */
+
+func (c *SHostDmesgCollector) Start() {
+	f, err := os.Open(kmsgPath)
+	if err != nil {
+		log.Errorf("failed open %s: %s", kmsgPath, err)
+		return
+	}
+	defer f.Close()
+
+	bootTime, err := getBootTime()
+	if err != nil {
+		log.Errorf("failed get boot time %s", err)
+		return
+	}
+	c.bootTime = bootTime
+
+	var currentBootStamp = bootTime.Unix()
+	var lastSeq = 0
+
+	readerState, err := c.loadState()
+	if err != nil {
+		log.Errorf("failed load readers state %s", err)
+	} else if readerState != nil {
+		if readerState.BootStamp == currentBootStamp {
+			lastSeq = readerState.LastSeq
+		}
+	}
+	log.Infof("Start dmesg reader from seq %d", lastSeq)
+
+	go func() {
+		for range time.Tick(flushInterval) {
+			c.mu.Lock()
+			c.flushBuffer()
+			c.mu.Unlock()
+		}
+	}()
+
+	reader := bufio.NewReader(f)
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			time.Sleep(100 * time.Millisecond)
+			continue
+		}
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+
+		entry, err := c.parseKmsgLine(line, bootTime)
+		if err != nil {
+			log.Errorf("failed parse kmsg line %s: %s", line, err)
+			continue
+		}
+		if entry.Seq <= lastSeq {
+			continue
+		}
+		// 只上传 warn 以上级别的日志
+		if entry.Level > 4 || c.isNoise(entry) {
+			continue
+		}
+
+		c.mu.Lock()
+		c.buffer = append(c.buffer, *entry)
+		if len(c.buffer) >= batchSize {
+			c.flushBuffer()
+		}
+		c.mu.Unlock()
+	}
+}
+
+func (c *SHostDmesgCollector) isNoise(entry *compute.SKmsgEntry) bool {
+	if strings.HasPrefix(entry.Message, "IPVS:") {
+		return true
+	}
+	return false
+}
+
+// flush buffer util success
+func (c *SHostDmesgCollector) flushBuffer() {
+	if len(c.buffer) == 0 {
+		return
+	}
+	seq := c.buffer[len(c.buffer)-1].Seq
+
+	for {
+		err := c.host.ReportHostDmesg(c.buffer)
+		if err != nil {
+			log.Errorf("failed report host dmesg %s", err)
+			time.Sleep(time.Second * 30)
+			continue
+		}
+		break
+	}
+
+	if err := c.saveState(seq); err != nil {
+		log.Errorf("failed save dmesg reader state: %s", err)
+	}
+
+	c.buffer = c.buffer[:0]
+}
+
+func (c *SHostDmesgCollector) loadState() (*ReaderState, error) {
+	dmesgStatePath := path.Join(filepath.Dir(options.HostOptions.ServersPath), "dmesg_reader_state")
+	if !fileutils2.Exists(dmesgStatePath) {
+		return nil, nil
+	}
+	data, err := fileutils2.FileGetContents(dmesgStatePath)
+	if err != nil {
+		return nil, err
+	}
+	jdata, err := jsonutils.ParseString(data)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed parse dmesg reader state")
+	}
+	var s ReaderState
+	err = jdata.Unmarshal(&s)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed unmarshal reader state")
+	}
+	return &s, nil
+}
+
+func (c *SHostDmesgCollector) saveState(seq int) error {
+	state := &ReaderState{
+		LastSeq:   seq,
+		BootStamp: c.bootTime.Unix(),
+	}
+	jstate := jsonutils.Marshal(state)
+	dmesgStatePath := path.Join(filepath.Dir(options.HostOptions.ServersPath), "dmesg_reader_state")
+	return fileutils2.FilePutContents(dmesgStatePath, jstate.String(), false)
+}
+
+type ReaderState struct {
+	LastSeq   int   `json:"last_seq"`
+	BootStamp int64 `json:"boot_stamp"` // UNIX seconds of boot time
+}
+
+func getBootTime() (time.Time, error) {
+	data, err := fileutils2.FileGetContents("/proc/uptime")
+	if err != nil {
+		return time.Time{}, err
+	}
+	fields := strings.Fields(data)
+	if len(fields) < 1 {
+		return time.Time{}, fmt.Errorf("invalid /proc/uptime")
+	}
+	uptimeSec, err := strconv.ParseFloat(fields[0], 64)
+	if err != nil {
+		return time.Time{}, err
+	}
+	return time.Now().Add(-time.Duration(uptimeSec * float64(time.Second))), nil
+}
+
+func (c *SHostDmesgCollector) parseKmsgLine(line string, bootTime time.Time) (*compute.SKmsgEntry, error) {
+	parts := strings.SplitN(line, ";", 2)
+	if len(parts) != 2 {
+		return nil, fmt.Errorf("invalid kmsg line: %s", line)
+	}
+
+	meta := strings.Split(parts[0], ",")
+	if len(meta) < 3 {
+		return nil, fmt.Errorf("invalid meta: %s", parts[0])
+	}
+
+	levelStr := strings.Trim(meta[0], "<>")
+	level, err := strconv.Atoi(levelStr)
+	if err != nil {
+		return nil, err
+	}
+
+	seq, _ := strconv.Atoi(meta[1])
+	timestamp, _ := strconv.ParseUint(meta[2], 10, 64)
+	rel := time.Duration(timestamp) * time.Microsecond
+	abs := bootTime.Add(rel)
+
+	return &compute.SKmsgEntry{
+		Level:   level,
+		Seq:     seq,
+		Message: parts[1],
+		Time:    abs,
+	}, nil
+}

pkg/hostman/hostmetrics/hostmetrics.go

@@ -41,6 +41,7 @@ import (
 	"yunion.io/x/onecloud/pkg/hostman/options"
 	"yunion.io/x/onecloud/pkg/util/fileutils2"
 	"yunion.io/x/onecloud/pkg/util/pod/stats"
+	"yunion.io/x/onecloud/pkg/util/timeutils2"
 )
 
 const (
@@ -63,18 +64,27 @@ type IHostInfo interface {
 	HasContainerVastaitechGpu() bool
 	HasContainerCphAmdGpu() bool
 	GetNvidiaGpuIndexMemoryMap() map[string]int
+	ReportHostDmesg(data []compute.SKmsgEntry) error
 }
 
+var hostDmesgCollector *SHostDmesgCollector
+
 func Init(hostInfo IHostInfo) {
 	if hostMetricsCollector == nil {
 		hostMetricsCollector = NewHostMetricsCollector(hostInfo)
 	}
+	if hostDmesgCollector == nil {
+		hostDmesgCollector = NewHostDmesgCollector(hostInfo)
+	}
 }
 
 func Start() {
 	if hostMetricsCollector != nil {
 		go hostMetricsCollector.Start()
 	}
+	if options.HostOptions.EnableDmesgCollect {
+		timeutils2.AddTimeout(30*time.Second, hostDmesgCollector.Start)
+	}
 }
 
 func Stop() {

pkg/hostman/options/options.go

@@ -45,6 +45,8 @@ type SHostBaseOptions struct {
 
 	DisableLocalVpc bool `help:"disable local VPC support" default:"false"`
 
+	EnableDmesgCollect bool `default:"true" help:"Enable dmesg collect or not, default true"`
+
 	DhcpLeaseTime   int `default:"100663296" help:"DHCP lease time in seconds"`
 	DhcpRenewalTime int `default:"67108864" help:"DHCP renewal time in seconds"`
 

pkg/mcclient/modules/mod_logs.go

@@ -33,6 +33,7 @@ var (
 	ActionLogs     modulebase.ResourceManager
 	CloudeventLogs modulebase.ResourceManager
 	ComputeLogs    modulebase.ResourceManager
+	DmesgLogs      modulebase.ResourceManager
 	MonitorLogs    modulebase.ResourceManager
 	NotifyLogs     modulebase.ResourceManager
 )
@@ -93,6 +94,9 @@ func init() {
 	ComputeLogs = NewComputeManager("event", "events",
 		[]string{"id", "ops_time", "obj_id", "obj_type", "obj_name", "user", "user_id", "tenant", "tenant_id", "owner_tenant_id", "action", "notes"},
 		[]string{})
+	DmesgLogs = NewComputeManager("hostdmesg", "hostdmesgs",
+		[]string{"id", "ops_time", "log_level", "obj_id", "obj_name", "user", "user_id", "tenant", "tenant_id", "owner_tenant_id", "notes"},
+		[]string{})
 	// ComputeLogs.SetApiVersion(mcclient.V2_API_VERSION)
 	MonitorLogs = NewMonitorV2Manager("event", "events",
 		[]string{"id", "ops_time", "obj_id", "obj_type", "obj_name", "user", "tenant", "action", "notes"},
@@ -103,4 +107,5 @@ func init() {
 
 	Logs = LogsManager{ComputeLogs}
 	Register(&Logs)
+	Register(&DmesgLogs)
 }