Go/Ocean
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add documentation

Browse Source
This commit is contained in:
Thorsten Sommer committed 2015-06-17 17:44:52 +02:00
1 parent 69d0bdabf4
commit 6dab89a1d4
144 files changed
+1107 -549
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Log/DeviceDatabase/ActivateLoggingDevice.go
+1
View File
@@ -4,6 +4,7 @@ import (
"github.com/SommerEngineering/Ocean/Log"
)
// Function with the setup of the logging device.
func ActivateLoggingDevice() {
Log.AddLoggingDevice(Database{})
}
Log/DeviceDatabase/CacheFull.go
+18
View File
@@ -0,0 +1,18 @@
package DeviceDatabase
// Function to check if the cache is full. If so, write all events to the database.
func cacheFull() {
mutexCacheFull.Lock()
defer mutexCacheFull.Unlock()
// Is the cache full?
if len(cache) < cacheSizeNumberOfEvents {
// Case: Cache is not full.
return
}
// Case: The cache is full. Write all events to the database.
for counter := 0; counter < cacheSizeNumberOfEvents; counter++ {
write2Database(<-cache)
}
}
Log/DeviceDatabase/CacheRefreshMessageNames.go
+37
View File
@@ -0,0 +1,37 @@
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Log"
LM "github.com/SommerEngineering/Ocean/Log/Meta"
"github.com/SommerEngineering/Ocean/Shutdown"
"time"
)
// Function for the thread which maintain the message name cache.
func cacheRefreshMessageNames() {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `The message names' refresh thread is now running.`)
// Create an own thread:
go func() {
// Endless loop:
for true {
// Read the message names rom the DB:
data := readMessageNamesFromDB()
mutexCacheMessageNames.Lock()
// Overwrite the cache:
cacheMessageNames = data
mutexCacheMessageNames.Unlock()
// Sleep some time:
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelTALKATIVE, LM.MessageNameEXECUTE, `The message names' cache was refreshed.`)
time.Sleep(time.Duration(nameCachesRefreshTimeSeconds) * time.Second)
// Case: The server goes down now.
if Shutdown.IsDown() {
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityLow, LM.ImpactLow, LM.MessageNameSHUTDOWN, `The message name's refresh thread is now shutting down.`)
return
}
}
}()
}
Log/DeviceDatabase/CacheRefreshSenderNames.go
+38
View File
@@ -0,0 +1,38 @@
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Log"
LM "github.com/SommerEngineering/Ocean/Log/Meta"
"github.com/SommerEngineering/Ocean/Shutdown"
"time"
)
// Function for the thread which maintain the sender name cache.
func cacheRefreshSenderNames() {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `The sender names' refresh thread is now running.`)
// Use an extra thread:
go func() {
// Endless lopp:
for true {
// Read the sender names from the DB:
data := readSenderNamesFromDB()
mutexCacheSenderNames.Lock()
// Overwrite the cache:
cacheSenderNames = data
mutexCacheSenderNames.Unlock()
// Sleep some time:
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelTALKATIVE, LM.MessageNameEXECUTE, `The sender names' cache was refreshed.`)
time.Sleep(time.Duration(nameCachesRefreshTimeSeconds) * time.Second)
// Case: The server is going down now.
if Shutdown.IsDown() {
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityLow, LM.ImpactLow, LM.MessageNameSHUTDOWN, `The sender name's refresh thread is now shutting down.`)
return
}
}
}()
}
Log/DeviceDatabase/Flush.go
+2
View File
@@ -1,5 +1,6 @@
package DeviceDatabase
// Flush the cache and write all messages to the database.
func (dev Database) Flush() {
mutexCacheFull.Lock()
defer mutexCacheFull.Unlock()
@@ -9,6 +10,7 @@ func (dev Database) Flush() {
write2Database(<-cache)
}
// Shutdown the database connection:
logDB.Logout()
logDBSession.Close()
}
Log/DeviceDatabase/Format.go
+3 -1
View File
@@ -4,8 +4,9 @@ import (
"github.com/SommerEngineering/Ocean/Log/Meta"
)
// Function to format a logging database entry as string.
func (entry LogDBEntry) Format() (result string) {
// First, we convert the logging db entry to the common logging type:
converted := Meta.Entry{}
converted.Time = entry.TimeUTC
converted.Project = entry.Project
@@ -18,6 +19,7 @@ func (entry LogDBEntry) Format() (result string) {
converted.MessageDescription = entry.MessageDescription
converted.Parameters = entry.Parameters
// Second, we can use then the format operation of these type:
result = converted.Format()
return
}
Log/DeviceDatabase/Init.go
+15 -1
View File
@@ -8,12 +8,19 @@ import (
"strconv"
)
// The init function for this package.
func init() {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `Starting now the database logging.`)
defer Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `Starting the database logging done.`)
// Init the database first:
initDatabase()
//
// Read all configuration values:
//
if value, err := strconv.Atoi(ConfigurationDB.Read(`LogDBCacheSizeNumberOfEvents`)); err != nil {
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityHigh, LM.ImpactUnknown, LM.MessageNameCONFIGURATION, `It was not possible to read the LogDBCacheSizeNumberOfEvents configuration.`, `The default value will be used.`, fmt.Sprintf(`Default value is %d.`, cacheSizeNumberOfEvents))
} else {
@@ -35,8 +42,15 @@ func init() {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameCONFIGURATION, `Configuration LogDBWebInterfaceNameCacheRefreshTimeSeconds was loaded.`, fmt.Sprintf(`The value is %d.`, nameCachesRefreshTimeSeconds))
}
// Create the cache:
cache = make(chan LogDBEntry, cacheSizeNumberOfEvents)
initTimeout()
// Starts a thread to write events based on time-outs:
startTimeout()
// Starts a thread to refresh the sender name cache:
cacheRefreshSenderNames()
// Starts a thread to refresh the message name cache:
cacheRefreshMessageNames()
}
Log/DeviceDatabase/InitDB.go
+10
View File
@@ -12,19 +12,26 @@ import (
"time"
)
// Init the database for the logging.
func initDatabase() {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameINIT, `Checking and init the logging database collection.`)
defer Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameINIT, `Checking and init the logging database collection done.`)
// Read the configuration values for the logging database:
databaseHost := ConfigurationDB.Read(`LogDBHost`)
databaseDB := ConfigurationDB.Read(`LogDBDatabase`)
databaseUsername := ConfigurationDB.Read(`LogDBUsername`)
databasePassword := ConfigurationDB.Read(`LogDBPassword`)
// Should the logging events at the database expire?
expire := strings.ToLower(ConfigurationDB.Read(`LogDBEventsExpire`)) == `true`
// The default values for the TTL (time to live):
expireAfterDays := 21900 // 60 years ~ maximum of MongoDB
expireValue4DisabledFunction := 21900 // 60 years ~ maximum of MongoDB
// Try to read the configured value for the TTL:
if value, errValue := strconv.Atoi(ConfigurationDB.Read(`LogDBEventsExpireAfterDays`)); errValue != nil {
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.SeverityMiddle, LM.ImpactUnknown, LM.MessageNameCONFIGURATION, `It was not possible to read the configuration for the expire time of logging events. Log events will not expire any more.`, errValue.Error())
expire = false
@@ -86,6 +93,9 @@ func initDatabase() {
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityUnknown, LM.ImpactUnknown, LM.MessageNameDATABASE, fmt.Sprintf(`Update the expire policy for the logging database done.`))
}
//
// Ensure that all necessary indexes are existing:
//
indexProject := mgo.Index{}
indexProject.Key = []string{`Project`}
logDBCollection.EnsureIndex(indexProject)
Log/DeviceDatabase/ReadCustom.go
+7 -1
View File
@@ -7,21 +7,27 @@ import (
func ReadCustom(timeRange, logLevel, logCategory, logImpact, logSeverity, logMessageName, logSender, logPage string) (events []LogDBEntry) {
//
// TODO => Is currently stub
// TODO => Is currently a stub
//
// Define the query:
query := logDBCollection.Find(bson.D{}).Sort(`-TimeUTC`).Limit(26)
count := 26
// Execute the query and count the results:
if n, err := query.Count(); err == nil {
count = n
}
// The iterator for the results:
iter := query.Iter()
entry := LogDBEntry{}
pos := 0
// Reserve the memory for the results:
events = make([]LogDBEntry, count)
// Loop over all entries and store it:
for iter.Next(&entry) {
events[pos] = entry
pos++
Log/DeviceDatabase/ReadLatest.go
+7 -1
View File
@@ -4,20 +4,26 @@ import (
"gopkg.in/mgo.v2/bson"
)
// Read the latest logging events from the database.
func ReadLatest() (events []LogDBEntry) {
// Define the query:
query := logDBCollection.Find(bson.D{}).Sort(`-TimeUTC`).Limit(26)
count := 26
// Execute the query and count the results:
if n, err := query.Count(); err == nil {
count = n
}
// The iterator for the results:
iter := query.Iter()
entry := LogDBEntry{}
pos := 0
// Reserve the memory for the results:
events = make([]LogDBEntry, count)
// Loop over all entries and store it:
for iter.Next(&entry) {
events[pos] = entry
pos++
Log/DeviceDatabase/ReadMessageNames.go
+1 -43
View File
@@ -1,51 +1,9 @@
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Log"
LM "github.com/SommerEngineering/Ocean/Log/Meta"
"github.com/SommerEngineering/Ocean/Shutdown"
"gopkg.in/mgo.v2/bson"
"sort"
"time"
)
// Read the message names out of the cache.
func ReadMessageNames() (messageNames []string) {
mutexCacheMessageNames.RLock()
defer mutexCacheMessageNames.RUnlock()
messageNames = cacheMessageNames
return
}
func cacheRefreshMessageNames() {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `The message names' refresh thread is now running.`)
go func() {
for true {
data := readMessageNamesFromDB()
mutexCacheMessageNames.Lock()
cacheMessageNames = data
mutexCacheMessageNames.Unlock()
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelTALKATIVE, LM.MessageNameEXECUTE, `The message names' cache was refreshed.`)
time.Sleep(time.Duration(nameCachesRefreshTimeSeconds) * time.Second)
if Shutdown.IsDown() {
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityLow, LM.ImpactLow, LM.MessageNameSHUTDOWN, `The message name's refresh thread is now shutting down.`)
return
}
}
}()
}
func readMessageNamesFromDB() (result []string) {
var nextMessageNames []string
if err := logDBCollection.Find(bson.D{}).Distinct(`MessageName`, &nextMessageNames); err != nil {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.MessageNameDATABASE, `Was not able to read the message names from the database.`, err.Error())
return
}
sort.Strings(nextMessageNames)
result = nextMessageNames
return
}
Log/DeviceDatabase/ReadMessageNamesFromDB.go
+23
View File
@@ -0,0 +1,23 @@
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Log"
LM "github.com/SommerEngineering/Ocean/Log/Meta"
"gopkg.in/mgo.v2/bson"
"sort"
)
// Read the message names from the database without any cache.
func readMessageNamesFromDB() (result []string) {
var nextMessageNames []string
if err := logDBCollection.Find(bson.D{}).Distinct(`MessageName`, &nextMessageNames); err != nil {
// Case: Error, was not able to write the event to the database:
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.MessageNameDATABASE, `Was not able to read the message names from the database.`, err.Error())
return
}
// Sort the sender names:
sort.Strings(nextMessageNames)
result = nextMessageNames
return
}
Log/DeviceDatabase/ReadSenderNames.go
+1 -43
View File
@@ -1,51 +1,9 @@
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Log"
LM "github.com/SommerEngineering/Ocean/Log/Meta"
"github.com/SommerEngineering/Ocean/Shutdown"
"gopkg.in/mgo.v2/bson"
"sort"
"time"
)
// Read the sender names out of the cache.
func ReadSenderNames() (senderNames []string) {
mutexCacheSenderNames.RLock()
defer mutexCacheSenderNames.RUnlock()
senderNames = cacheSenderNames
return
}
func cacheRefreshSenderNames() {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelINFO, LM.MessageNameSTARTUP, `The sender names' refresh thread is now running.`)
go func() {
for true {
data := readSenderNamesFromDB()
mutexCacheSenderNames.Lock()
cacheSenderNames = data
mutexCacheSenderNames.Unlock()
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelTALKATIVE, LM.MessageNameEXECUTE, `The sender names' cache was refreshed.`)
time.Sleep(time.Duration(nameCachesRefreshTimeSeconds) * time.Second)
if Shutdown.IsDown() {
Log.LogFull(senderName, LM.CategorySYSTEM, LM.LevelWARN, LM.SeverityLow, LM.ImpactLow, LM.MessageNameSHUTDOWN, `The sender name's refresh thread is now shutting down.`)
return
}
}
}()
}
func readSenderNamesFromDB() (result []string) {
var nextSenderNames []string
if err := logDBCollection.Find(bson.D{}).Distinct(`Sender`, &nextSenderNames); err != nil {
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.MessageNameDATABASE, `Was not able to read the sender names from the database.`, err.Error())
return
}
sort.Strings(nextSenderNames)
result = nextSenderNames
return
}
Log/DeviceDatabase/ReadSenderNamesFromDB.go
+23
View File
@@ -0,0 +1,23 @@
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Log"
LM "github.com/SommerEngineering/Ocean/Log/Meta"
"gopkg.in/mgo.v2/bson"
"sort"
)
// Reads the sender names from the database without any caching.
func readSenderNamesFromDB() (result []string) {
var nextSenderNames []string
if err := logDBCollection.Find(bson.D{}).Distinct(`Sender`, &nextSenderNames); err != nil {
// Case: Was not possible to write to the database.
Log.LogShort(senderName, LM.CategorySYSTEM, LM.LevelERROR, LM.MessageNameDATABASE, `Was not able to read the sender names from the database.`, err.Error())
return
}
// Sort the sender names:
sort.Strings(nextSenderNames)
result = nextSenderNames
return
}
Log/DeviceDatabase/ReceiveLogging.go
+2 -3
View File
@@ -4,14 +4,13 @@ import (
"github.com/SommerEngineering/Ocean/Log/Meta"
)
type Database struct {
}
// This function is the interface between the logging system and the database logger.
func (dev Database) Log(entries []Meta.Entry) {
//
// Cannot log here to prevent endless loop (consumer is also producer)
//
// Write every incoming batch to the cache:
write2Cache(entries)
}
Log/DeviceDatabase/Scheme.go
+6
View File
@@ -4,6 +4,7 @@ import (
"time"
)
// The type for the database logging.
type LogDBEntry struct {
TimeUTC time.Time `bson:"TimeUTC"`
Project string `bson:"Project"`
@@ -17,8 +18,13 @@ type LogDBEntry struct {
Parameters []string `bson:"Parameters"`
}
// A type for the TTL (time to live) for the index.
type TTLUpdateResult struct {
ExpireAfterSeconds_old int32 `bson:"expireAfterSeconds_old"`
ExpireAfterSeconds_new int32 `bson:"expireAfterSeconds_new"`
Ok int32 `bson:"ok"`
}
// The logging device.
type Database struct {
}
Log/DeviceDatabase/ReadFromCache.go → Log/DeviceDatabase/StartTimeout.go
+32 -41
View File
@@ -1,41 +1,32 @@
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Shutdown"
"time"
)
// Case: The cache is full
func cacheFull() {
mutexCacheFull.Lock()
defer mutexCacheFull.Unlock()
if len(cache) < cacheSizeNumberOfEvents {
return
}
for counter := 0; counter < cacheSizeNumberOfEvents; counter++ {
write2Database(<-cache)
}
}
// Case: Time out
func initTimeout() {
go func() {
for {
if Shutdown.IsDown() {
return
}
time.Sleep(time.Duration(cacheSizeTime2FlushSeconds) * time.Second)
mutexCacheFull.Lock()
amount := len(cache)
for counter := 0; counter < amount; counter++ {
write2Database(<-cache)
}
mutexCacheFull.Unlock()
}
}()
}
package DeviceDatabase
import (
"github.com/SommerEngineering/Ocean/Shutdown"
"time"
)
// The timeout function writes the logging events to the database afer some time.
func startTimeout() {
// Starts a new thread:
go func() {
for {
// Case: The system goes down now.
if Shutdown.IsDown() {
return
}
// Wait for the right time:
time.Sleep(time.Duration(cacheSizeTime2FlushSeconds) * time.Second)
// Write the events to the database:
mutexCacheFull.Lock()
amount := len(cache)
for counter := 0; counter < amount; counter++ {
write2Database(<-cache)
}
mutexCacheFull.Unlock()
}
}()
}
Log/DeviceDatabase/Variables.go
+1
View File
@@ -7,6 +7,7 @@ import (
)
var (
// This is the name for logging event from this package:
senderName LM.Sender = `System::Logger::Database`
mutexCacheFull sync.Mutex = sync.Mutex{}
mutexCacheSenderNames sync.RWMutex = sync.RWMutex{}
Log/DeviceDatabase/Write2Cache.go
+7
View File
@@ -4,12 +4,17 @@ import (
"github.com/SommerEngineering/Ocean/Log/Meta"
)
// This function writes a batch of log entries to the cache.
func write2Cache(entries []Meta.Entry) {
// Loop over each entry:
for _, entry := range entries {
// If the cache is full, execute it:
if len(cache) == cacheSizeNumberOfEvents {
// Execute the cache with a new thread:
go cacheFull()
}
// Convert the log entry to the database format:
logDBentry := LogDBEntry{}
logDBentry.Category = entry.Category.Format()
logDBentry.Impact = entry.Impact.Format()
@@ -21,6 +26,8 @@ func write2Cache(entries []Meta.Entry) {
logDBentry.Sender = string(entry.Sender)
logDBentry.Severity = entry.Severity.Format()
logDBentry.TimeUTC = entry.Time
// Write it to the cache:
cache <- logDBentry
}
}
Log/DeviceDatabase/Write2Database.go
+9 -1
View File
@@ -1,7 +1,15 @@
package DeviceDatabase
import (
"fmt"
)
// Function to write a logging event to the database.
func write2Database(entry LogDBEntry) {
// Try to write the event to the database:
if err := logDBCollection.Insert(entry); err != nil {
// Can not log here to prevent endless loop (consumer is also producer)
// Case: Error!
// Cannot log here to prevent endless loop (consumer is also producer)
fmt.Printf("Was not able to write a logging event to the database: '%s'. The log entry was: '%s'.\n", err.Error(), entry.Format())
}
}