fluent-logger-R/fluent-logger-R.R at master · ciela/fluent-logger-R · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
library("rjson")

FluentLogger <- setRefClass(
  "FluentLogger",

  fields=c("tag", "host", "port", "timeout", "sender"),

  methods=list(

    post=function(label, data, timestamp=as.integer(Sys.time())) {
      tryCatch({
          jsondata <- toJSON(list(paste(tag, label, sep="."), timestamp, data))
          writeLines(jsondata, sender)
          .self$flush()
          invisible(TRUE)
        },
        warning=function(w) { message(w); invisible(FALSE) },
        error=function(e) { message(e); invisible(FALSE) }
      )
    },

    close=function() {
      tryCatch({
          if (isOpen(sender)) {
            close.connection(sender)
            sender <<- NULL

            # delete from secret object .loggers
            loggers <- get(".loggers", envir=FluentLoggerEnv)
            loggers[[paste(tag, host, port, timeout, sep="_")]] <- NULL
            assign(".loggers", loggers, envir=FluentLoggerEnv)
          }
        },
        warning=function(w) { message(w) }, # ignore
        error=function(e) { message(e) } # ignore
      )
    },

    flush=function() {
      tryCatch({
          if (isIncomplete(sender)) {
            flush.connection(sender)
          }
        },
        warning=function(w) { message(w) }, # ignore
        error=function(e) { message(e) } # ignore
      )
    }
  )
)

getFluentLogger <- function(tag, host, port=24224, timeout=3) {

  # if there aren't exist the environment FluentLoggerEnv
  if (!exists("FluentLoggerEnv", envir=.GlobalEnv)) {
    # make the envir.
    FluentLoggerEnv <- new.env(parent=.GlobalEnv)
    assign("FluentLoggerEnv", FluentLoggerEnv, envir=.GlobalEnv)
  }

  # get secret object .loggers from FluentLoggerEnv
  if (exists(".loggers", envir=FluentLoggerEnv)) {
    loggers <- get(".loggers", envir=FluentLoggerEnv)
  } else {
    loggers <- list()
  }

  # search logger by tag, host, port and timeout.
  key <- paste(tag, host, port, timeout, sep="_")
  if (is.null(loggers[[key]])) {
    # if loggers doesn't contain the key, make new FluentLogger.
    stopF <- function(e) stop(paste("Failed to connect fluentd:", paste(host, port, sep=":")), call.=F)
    tryCatch(
      sender <- socketConnection(host, port, timeout=timeout),
      warning=stopF,
      error=stopF
    )
    loggers[[key]] <- FluentLogger$new(tag=tag, host=host, port=port, timeout=timeout, sender=sender)
    assign(".loggers", loggers, envir=FluentLoggerEnv)
  }

  loggers[[key]]
}

closeAllFluentLoggers <- function() {
  if (exists("FluentLoggerEnv", envir=.GlobalEnv) && exists(".loggers", envir=FluentLoggerEnv)) {
    loggers <- get(".loggers", envir=FluentLoggerEnv)
    for (logger in loggers) {
      logger$close()
      logger <- NULL
    }
    # remove .loggers from FluentLoggerEnv
    rm(".loggers", envir=FluentLoggerEnv)
  }
}

flushAllFluentLoggers <- function() {
  if (exists("FluentLoggerEnv", envir=.GlobalEnv) && exists(".loggers", envir=FluentLoggerEnv)) {
    loggers <- get(".loggers", envir=FluentLoggerEnv)
    for (logger in loggers) {
      logger$flush()
    }
  }
}