Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

finished changing the logs format to include the tenant and user name #1012

Merged
merged 5 commits into from
Jun 21, 2023
Merged

finished changing the logs format to include the tenant and user name #1012

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
61d59e2
finished changing the logs format to include the tenant and user name
Jun 19, 2023
4a59160
resolved conflicts
Jun 19, 2023
dd0a831
shortened the logs prefix and made the logs more accurate
Jun 20, 2023
a00e4ca
resolved comments
Jun 21, 2023
208fa57
resolved conflicts
Jun 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 21 additions & 21 deletions server/background_tasks.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ func (s *Server) ListenForZombieConnCheckRequests() error {
if len(connectionIds) > 0 { // in case there are connections
bytes, err := json.Marshal(connectionIds)
if err != nil {
s.Errorf("ListenForZombieConnCheckRequests: " + err.Error())
s.Errorf("ListenForZombieConnCheckRequests: %v", err.Error())
} else {
s.sendInternalAccountMsgWithReply(s.GlobalAccount(), reply, _EMPTY_, nil, bytes, true)
}
Expand All @@ -73,7 +73,7 @@ func (s *Server) ListenForIntegrationsUpdateEvents() error {
var integrationUpdate models.CreateIntegrationSchema
err := json.Unmarshal(msg, &integrationUpdate)
if err != nil {
s.Errorf("ListenForIntegrationsUpdateEvents: " + err.Error())
s.Errorf("[tenant: %v]ListenForIntegrationsUpdateEvents: %v", integrationUpdate.TenantName, err.Error())
return
}
switch strings.ToLower(integrationUpdate.Name) {
Expand All @@ -85,7 +85,7 @@ func (s *Server) ListenForIntegrationsUpdateEvents() error {
case "s3":
CacheDetails("s3", integrationUpdate.Keys, integrationUpdate.Properties, integrationUpdate.TenantName)
default:
s.Warnf("ListenForIntegrationsUpdateEvents: %s %s", strings.ToLower(integrationUpdate.Name), "unknown integration")
s.Warnf("[tenant: %v] ListenForIntegrationsUpdateEvents: %s %s", integrationUpdate.TenantName, strings.ToLower(integrationUpdate.Name), "unknown integration")
return
}
}(copyBytes(msg))
Expand All @@ -102,7 +102,7 @@ func (s *Server) ListenForConfigReloadEvents() error {
// reload config
err := s.Reload()
if err != nil {
s.Errorf("Failed reloading: " + err.Error())
s.Errorf("Failed reloading: %v", err.Error())
}
}(copyBytes(msg))
})
Expand All @@ -117,13 +117,13 @@ func (s *Server) ListenForNotificationEvents() error {
go func(msg []byte) {
tenantName, message, err := s.getTenantNameAndMessage(msg)
if err != nil {
s.Errorf("ListenForNotificationEvents: " + err.Error())
s.Errorf("[tenant: %v]ListenForNotificationEvents: %v", tenantName, err.Error())
return
}
var notification models.Notification
err = json.Unmarshal([]byte(message), &notification)
if err != nil {
s.Errorf("ListenForNotificationEvents: " + err.Error())
s.Errorf("[tenant: %v]ListenForNotificationEvents: %v", tenantName, err.Error())
return
}
notificationMsg := notification.Msg
Expand All @@ -147,13 +147,13 @@ func (s *Server) ListenForPoisonMsgAcks() error {
go func(msg []byte) {
tenantName, message, err := s.getTenantNameAndMessage(msg)
if err != nil {
s.Errorf("ListenForPoisonMsgAcks: " + err.Error())
s.Errorf("[tenant: %v]ListenForPoisonMsgAcks: %v", tenantName, err.Error())
return
}
var msgToAck models.PmAckMsg
err = json.Unmarshal([]byte(message), &msgToAck)
if err != nil {
s.Errorf("ListenForPoisonMsgAcks: " + err.Error())
s.Errorf("[tenant: %v]ListenForPoisonMsgAcks: %v", tenantName, err.Error())
return
}
err = db.RemoveCgFromDlsMsg(msgToAck.ID, msgToAck.CgName, tenantName)
Expand Down Expand Up @@ -259,7 +259,7 @@ func (s *Server) StartBackgroundTasks() error {
go s.InitializeThroughputSampling()
go s.UploadTenantUsageToDB()
go s.RefreshFirebaseFunctionsKey()

return nil
}

Expand All @@ -282,15 +282,15 @@ func (s *Server) uploadMsgsToTier2Storage() {
}
err := serv.memphisAddConsumer(globalAccountName, tieredStorageStream, &cc)
if err != nil {
serv.Errorf("Failed add tiered storage consumer: " + err.Error())
serv.Errorf("Failed add tiered storage consumer: %v", err.Error())
return
}
TIERED_STORAGE_CONSUMER_CREATED = true
}
tieredStorageMapLock.Lock()
err := flushMapToTire2Storage()
if err != nil {
serv.Errorf("Failed upload messages to tiered 2 storage: " + err.Error())
serv.Errorf("Failed upload messages to tiered 2 storage: %v", err.Error())
tieredStorageMapLock.Unlock()
continue
}
Expand Down Expand Up @@ -336,7 +336,7 @@ func (s *Server) ConsumeUnackedMsgs() {
}(subject, reply, copyBytes(msg))
})
if err != nil {
s.Errorf("Failed to subscribe to unacked messages: " + err.Error())
s.Errorf("Failed to subscribe to unacked messages: %v", err.Error())
continue
}

Expand Down Expand Up @@ -403,7 +403,7 @@ func (s *Server) ConsumeTieredStorageMsgs() {
}(subject, reply, copyBytes(msg))
})
if err != nil {
s.Errorf("Failed to subscribe to tiered storage messages: " + err.Error())
s.Errorf("Failed to subscribe to tiered storage messages: %v", err.Error())
continue
}

Expand Down Expand Up @@ -445,41 +445,41 @@ func (s *Server) ListenForSchemaverseDlsEvents() error {
go func(msg []byte) {
tenantName, stringMessage, err := s.getTenantNameAndMessage(msg)
if err != nil {
s.Errorf("ListenForNotificationEvents: " + err.Error())
s.Errorf("[tenant: %v]ListenForNotificationEvents: %v", tenantName, err.Error())
return
}
var message models.SchemaVerseDlsMessageSdk
err = json.Unmarshal([]byte(stringMessage), &message)
if err != nil {
serv.Errorf("ListenForSchemaverseDlsEvents: " + err.Error())
serv.Errorf("[tenant: %v]ListenForSchemaverseDlsEvents: %v", tenantName, err.Error())
return
}

exist, station, err := db.GetStationByName(message.StationName, tenantName)
if err != nil {
serv.Errorf("ListenForSchemaverseDlsEvents: " + err.Error())
serv.Errorf("[tenant: %v]ListenForSchemaverseDlsEvents: %v", tenantName, err.Error())
return
}
if !exist {
serv.Warnf("ListenForSchemaverseDlsEvents: station " + message.StationName + " couldn't been found")
serv.Warnf("[tenant: %v]ListenForSchemaverseDlsEvents: station %v couldn't been found", tenantName, message.StationName)
return
}

exist, p, err := db.GetProducerByNameAndConnectionID(message.Producer.Name, message.Producer.ConnectionId)
if err != nil {
serv.Errorf("ListenForSchemaverseDlsEvents: " + err.Error())
serv.Errorf("[tenant: %v]ListenForSchemaverseDlsEvents: %v", tenantName, err.Error())
return
}

if !exist {
serv.Warnf("ListenForSchemaverseDlsEvents: producer " + p.Name + " couldn't been found")
serv.Warnf("[tenant: %v]ListenForSchemaverseDlsEvents: producer %v couldn't been found", tenantName, p.Name)
return
}

message.Message.TimeSent = time.Now()
_, err = db.InsertSchemaverseDlsMsg(station.ID, 0, p.ID, []string{}, models.MessagePayload(message.Message), message.ValidationError, tenantName)
if err != nil {
serv.Errorf("ListenForSchemaverseDlsEvents: " + err.Error())
serv.Errorf("[tenant: %v]ListenForSchemaverseDlsEvents: %v", tenantName, err.Error())
return
}
}(copyBytes(msg))
Expand All @@ -497,7 +497,7 @@ func (s *Server) RemoveOldDlsMsgs() {
configurationTime := time.Now().Add(time.Hour * time.Duration(-s.opts.DlsRetentionHours))
err := db.DeleteOldDlsMessageByRetention(configurationTime)
if err != nil {
serv.Errorf("RemoveOldDlsMsgs: " + err.Error())
serv.Errorf("RemoveOldDlsMsgs: %v", err.Error())
}
}
}
Loading