You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafficcontrol.apache.org by ne...@apache.org on 2017/01/30 15:29:16 UTC
[10/19] incubator-trafficcontrol git commit: Move TM2 to
trafficcontrol/traffic_monitor_golang
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor/experimental/vendor/gopkg.in/fsnotify.v1/windows.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/vendor/gopkg.in/fsnotify.v1/windows.go b/traffic_monitor/experimental/vendor/gopkg.in/fsnotify.v1/windows.go
deleted file mode 100644
index c836bdb..0000000
--- a/traffic_monitor/experimental/vendor/gopkg.in/fsnotify.v1/windows.go
+++ /dev/null
@@ -1,561 +0,0 @@
-// Copyright 2011 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-// +build windows
-
-package fsnotify
-
-import (
- "errors"
- "fmt"
- "os"
- "path/filepath"
- "runtime"
- "sync"
- "syscall"
- "unsafe"
-)
-
-// Watcher watches a set of files, delivering events to a channel.
-type Watcher struct {
- Events chan Event
- Errors chan error
- isClosed bool // Set to true when Close() is first called
- mu sync.Mutex // Map access
- port syscall.Handle // Handle to completion port
- watches watchMap // Map of watches (key: i-number)
- input chan *input // Inputs to the reader are sent on this channel
- quit chan chan<- error
-}
-
-// NewWatcher establishes a new watcher with the underlying OS and begins waiting for events.
-func NewWatcher() (*Watcher, error) {
- port, e := syscall.CreateIoCompletionPort(syscall.InvalidHandle, 0, 0, 0)
- if e != nil {
- return nil, os.NewSyscallError("CreateIoCompletionPort", e)
- }
- w := &Watcher{
- port: port,
- watches: make(watchMap),
- input: make(chan *input, 1),
- Events: make(chan Event, 50),
- Errors: make(chan error),
- quit: make(chan chan<- error, 1),
- }
- go w.readEvents()
- return w, nil
-}
-
-// Close removes all watches and closes the events channel.
-func (w *Watcher) Close() error {
- if w.isClosed {
- return nil
- }
- w.isClosed = true
-
- // Send "quit" message to the reader goroutine
- ch := make(chan error)
- w.quit <- ch
- if err := w.wakeupReader(); err != nil {
- return err
- }
- return <-ch
-}
-
-// Add starts watching the named file or directory (non-recursively).
-func (w *Watcher) Add(name string) error {
- if w.isClosed {
- return errors.New("watcher already closed")
- }
- in := &input{
- op: opAddWatch,
- path: filepath.Clean(name),
- flags: sysFSALLEVENTS,
- reply: make(chan error),
- }
- w.input <- in
- if err := w.wakeupReader(); err != nil {
- return err
- }
- return <-in.reply
-}
-
-// Remove stops watching the the named file or directory (non-recursively).
-func (w *Watcher) Remove(name string) error {
- in := &input{
- op: opRemoveWatch,
- path: filepath.Clean(name),
- reply: make(chan error),
- }
- w.input <- in
- if err := w.wakeupReader(); err != nil {
- return err
- }
- return <-in.reply
-}
-
-const (
- // Options for AddWatch
- sysFSONESHOT = 0x80000000
- sysFSONLYDIR = 0x1000000
-
- // Events
- sysFSACCESS = 0x1
- sysFSALLEVENTS = 0xfff
- sysFSATTRIB = 0x4
- sysFSCLOSE = 0x18
- sysFSCREATE = 0x100
- sysFSDELETE = 0x200
- sysFSDELETESELF = 0x400
- sysFSMODIFY = 0x2
- sysFSMOVE = 0xc0
- sysFSMOVEDFROM = 0x40
- sysFSMOVEDTO = 0x80
- sysFSMOVESELF = 0x800
-
- // Special events
- sysFSIGNORED = 0x8000
- sysFSQOVERFLOW = 0x4000
-)
-
-func newEvent(name string, mask uint32) Event {
- e := Event{Name: name}
- if mask&sysFSCREATE == sysFSCREATE || mask&sysFSMOVEDTO == sysFSMOVEDTO {
- e.Op |= Create
- }
- if mask&sysFSDELETE == sysFSDELETE || mask&sysFSDELETESELF == sysFSDELETESELF {
- e.Op |= Remove
- }
- if mask&sysFSMODIFY == sysFSMODIFY {
- e.Op |= Write
- }
- if mask&sysFSMOVE == sysFSMOVE || mask&sysFSMOVESELF == sysFSMOVESELF || mask&sysFSMOVEDFROM == sysFSMOVEDFROM {
- e.Op |= Rename
- }
- if mask&sysFSATTRIB == sysFSATTRIB {
- e.Op |= Chmod
- }
- return e
-}
-
-const (
- opAddWatch = iota
- opRemoveWatch
-)
-
-const (
- provisional uint64 = 1 << (32 + iota)
-)
-
-type input struct {
- op int
- path string
- flags uint32
- reply chan error
-}
-
-type inode struct {
- handle syscall.Handle
- volume uint32
- index uint64
-}
-
-type watch struct {
- ov syscall.Overlapped
- ino *inode // i-number
- path string // Directory path
- mask uint64 // Directory itself is being watched with these notify flags
- names map[string]uint64 // Map of names being watched and their notify flags
- rename string // Remembers the old name while renaming a file
- buf [4096]byte
-}
-
-type indexMap map[uint64]*watch
-type watchMap map[uint32]indexMap
-
-func (w *Watcher) wakeupReader() error {
- e := syscall.PostQueuedCompletionStatus(w.port, 0, 0, nil)
- if e != nil {
- return os.NewSyscallError("PostQueuedCompletionStatus", e)
- }
- return nil
-}
-
-func getDir(pathname string) (dir string, err error) {
- attr, e := syscall.GetFileAttributes(syscall.StringToUTF16Ptr(pathname))
- if e != nil {
- return "", os.NewSyscallError("GetFileAttributes", e)
- }
- if attr&syscall.FILE_ATTRIBUTE_DIRECTORY != 0 {
- dir = pathname
- } else {
- dir, _ = filepath.Split(pathname)
- dir = filepath.Clean(dir)
- }
- return
-}
-
-func getIno(path string) (ino *inode, err error) {
- h, e := syscall.CreateFile(syscall.StringToUTF16Ptr(path),
- syscall.FILE_LIST_DIRECTORY,
- syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE|syscall.FILE_SHARE_DELETE,
- nil, syscall.OPEN_EXISTING,
- syscall.FILE_FLAG_BACKUP_SEMANTICS|syscall.FILE_FLAG_OVERLAPPED, 0)
- if e != nil {
- return nil, os.NewSyscallError("CreateFile", e)
- }
- var fi syscall.ByHandleFileInformation
- if e = syscall.GetFileInformationByHandle(h, &fi); e != nil {
- syscall.CloseHandle(h)
- return nil, os.NewSyscallError("GetFileInformationByHandle", e)
- }
- ino = &inode{
- handle: h,
- volume: fi.VolumeSerialNumber,
- index: uint64(fi.FileIndexHigh)<<32 | uint64(fi.FileIndexLow),
- }
- return ino, nil
-}
-
-// Must run within the I/O thread.
-func (m watchMap) get(ino *inode) *watch {
- if i := m[ino.volume]; i != nil {
- return i[ino.index]
- }
- return nil
-}
-
-// Must run within the I/O thread.
-func (m watchMap) set(ino *inode, watch *watch) {
- i := m[ino.volume]
- if i == nil {
- i = make(indexMap)
- m[ino.volume] = i
- }
- i[ino.index] = watch
-}
-
-// Must run within the I/O thread.
-func (w *Watcher) addWatch(pathname string, flags uint64) error {
- dir, err := getDir(pathname)
- if err != nil {
- return err
- }
- if flags&sysFSONLYDIR != 0 && pathname != dir {
- return nil
- }
- ino, err := getIno(dir)
- if err != nil {
- return err
- }
- w.mu.Lock()
- watchEntry := w.watches.get(ino)
- w.mu.Unlock()
- if watchEntry == nil {
- if _, e := syscall.CreateIoCompletionPort(ino.handle, w.port, 0, 0); e != nil {
- syscall.CloseHandle(ino.handle)
- return os.NewSyscallError("CreateIoCompletionPort", e)
- }
- watchEntry = &watch{
- ino: ino,
- path: dir,
- names: make(map[string]uint64),
- }
- w.mu.Lock()
- w.watches.set(ino, watchEntry)
- w.mu.Unlock()
- flags |= provisional
- } else {
- syscall.CloseHandle(ino.handle)
- }
- if pathname == dir {
- watchEntry.mask |= flags
- } else {
- watchEntry.names[filepath.Base(pathname)] |= flags
- }
- if err = w.startRead(watchEntry); err != nil {
- return err
- }
- if pathname == dir {
- watchEntry.mask &= ^provisional
- } else {
- watchEntry.names[filepath.Base(pathname)] &= ^provisional
- }
- return nil
-}
-
-// Must run within the I/O thread.
-func (w *Watcher) remWatch(pathname string) error {
- dir, err := getDir(pathname)
- if err != nil {
- return err
- }
- ino, err := getIno(dir)
- if err != nil {
- return err
- }
- w.mu.Lock()
- watch := w.watches.get(ino)
- w.mu.Unlock()
- if watch == nil {
- return fmt.Errorf("can't remove non-existent watch for: %s", pathname)
- }
- if pathname == dir {
- w.sendEvent(watch.path, watch.mask&sysFSIGNORED)
- watch.mask = 0
- } else {
- name := filepath.Base(pathname)
- w.sendEvent(watch.path+"\\"+name, watch.names[name]&sysFSIGNORED)
- delete(watch.names, name)
- }
- return w.startRead(watch)
-}
-
-// Must run within the I/O thread.
-func (w *Watcher) deleteWatch(watch *watch) {
- for name, mask := range watch.names {
- if mask&provisional == 0 {
- w.sendEvent(watch.path+"\\"+name, mask&sysFSIGNORED)
- }
- delete(watch.names, name)
- }
- if watch.mask != 0 {
- if watch.mask&provisional == 0 {
- w.sendEvent(watch.path, watch.mask&sysFSIGNORED)
- }
- watch.mask = 0
- }
-}
-
-// Must run within the I/O thread.
-func (w *Watcher) startRead(watch *watch) error {
- if e := syscall.CancelIo(watch.ino.handle); e != nil {
- w.Errors <- os.NewSyscallError("CancelIo", e)
- w.deleteWatch(watch)
- }
- mask := toWindowsFlags(watch.mask)
- for _, m := range watch.names {
- mask |= toWindowsFlags(m)
- }
- if mask == 0 {
- if e := syscall.CloseHandle(watch.ino.handle); e != nil {
- w.Errors <- os.NewSyscallError("CloseHandle", e)
- }
- w.mu.Lock()
- delete(w.watches[watch.ino.volume], watch.ino.index)
- w.mu.Unlock()
- return nil
- }
- e := syscall.ReadDirectoryChanges(watch.ino.handle, &watch.buf[0],
- uint32(unsafe.Sizeof(watch.buf)), false, mask, nil, &watch.ov, 0)
- if e != nil {
- err := os.NewSyscallError("ReadDirectoryChanges", e)
- if e == syscall.ERROR_ACCESS_DENIED && watch.mask&provisional == 0 {
- // Watched directory was probably removed
- if w.sendEvent(watch.path, watch.mask&sysFSDELETESELF) {
- if watch.mask&sysFSONESHOT != 0 {
- watch.mask = 0
- }
- }
- err = nil
- }
- w.deleteWatch(watch)
- w.startRead(watch)
- return err
- }
- return nil
-}
-
-// readEvents reads from the I/O completion port, converts the
-// received events into Event objects and sends them via the Events channel.
-// Entry point to the I/O thread.
-func (w *Watcher) readEvents() {
- var (
- n, key uint32
- ov *syscall.Overlapped
- )
- runtime.LockOSThread()
-
- for {
- e := syscall.GetQueuedCompletionStatus(w.port, &n, &key, &ov, syscall.INFINITE)
- watch := (*watch)(unsafe.Pointer(ov))
-
- if watch == nil {
- select {
- case ch := <-w.quit:
- w.mu.Lock()
- var indexes []indexMap
- for _, index := range w.watches {
- indexes = append(indexes, index)
- }
- w.mu.Unlock()
- for _, index := range indexes {
- for _, watch := range index {
- w.deleteWatch(watch)
- w.startRead(watch)
- }
- }
- var err error
- if e := syscall.CloseHandle(w.port); e != nil {
- err = os.NewSyscallError("CloseHandle", e)
- }
- close(w.Events)
- close(w.Errors)
- ch <- err
- return
- case in := <-w.input:
- switch in.op {
- case opAddWatch:
- in.reply <- w.addWatch(in.path, uint64(in.flags))
- case opRemoveWatch:
- in.reply <- w.remWatch(in.path)
- }
- default:
- }
- continue
- }
-
- switch e {
- case syscall.ERROR_MORE_DATA:
- if watch == nil {
- w.Errors <- errors.New("ERROR_MORE_DATA has unexpectedly null lpOverlapped buffer")
- } else {
- // The i/o succeeded but the buffer is full.
- // In theory we should be building up a full packet.
- // In practice we can get away with just carrying on.
- n = uint32(unsafe.Sizeof(watch.buf))
- }
- case syscall.ERROR_ACCESS_DENIED:
- // Watched directory was probably removed
- w.sendEvent(watch.path, watch.mask&sysFSDELETESELF)
- w.deleteWatch(watch)
- w.startRead(watch)
- continue
- case syscall.ERROR_OPERATION_ABORTED:
- // CancelIo was called on this handle
- continue
- default:
- w.Errors <- os.NewSyscallError("GetQueuedCompletionPort", e)
- continue
- case nil:
- }
-
- var offset uint32
- for {
- if n == 0 {
- w.Events <- newEvent("", sysFSQOVERFLOW)
- w.Errors <- errors.New("short read in readEvents()")
- break
- }
-
- // Point "raw" to the event in the buffer
- raw := (*syscall.FileNotifyInformation)(unsafe.Pointer(&watch.buf[offset]))
- buf := (*[syscall.MAX_PATH]uint16)(unsafe.Pointer(&raw.FileName))
- name := syscall.UTF16ToString(buf[:raw.FileNameLength/2])
- fullname := watch.path + "\\" + name
-
- var mask uint64
- switch raw.Action {
- case syscall.FILE_ACTION_REMOVED:
- mask = sysFSDELETESELF
- case syscall.FILE_ACTION_MODIFIED:
- mask = sysFSMODIFY
- case syscall.FILE_ACTION_RENAMED_OLD_NAME:
- watch.rename = name
- case syscall.FILE_ACTION_RENAMED_NEW_NAME:
- if watch.names[watch.rename] != 0 {
- watch.names[name] |= watch.names[watch.rename]
- delete(watch.names, watch.rename)
- mask = sysFSMOVESELF
- }
- }
-
- sendNameEvent := func() {
- if w.sendEvent(fullname, watch.names[name]&mask) {
- if watch.names[name]&sysFSONESHOT != 0 {
- delete(watch.names, name)
- }
- }
- }
- if raw.Action != syscall.FILE_ACTION_RENAMED_NEW_NAME {
- sendNameEvent()
- }
- if raw.Action == syscall.FILE_ACTION_REMOVED {
- w.sendEvent(fullname, watch.names[name]&sysFSIGNORED)
- delete(watch.names, name)
- }
- if w.sendEvent(fullname, watch.mask&toFSnotifyFlags(raw.Action)) {
- if watch.mask&sysFSONESHOT != 0 {
- watch.mask = 0
- }
- }
- if raw.Action == syscall.FILE_ACTION_RENAMED_NEW_NAME {
- fullname = watch.path + "\\" + watch.rename
- sendNameEvent()
- }
-
- // Move to the next event in the buffer
- if raw.NextEntryOffset == 0 {
- break
- }
- offset += raw.NextEntryOffset
-
- // Error!
- if offset >= n {
- w.Errors <- errors.New("Windows system assumed buffer larger than it is, events have likely been missed.")
- break
- }
- }
-
- if err := w.startRead(watch); err != nil {
- w.Errors <- err
- }
- }
-}
-
-func (w *Watcher) sendEvent(name string, mask uint64) bool {
- if mask == 0 {
- return false
- }
- event := newEvent(name, uint32(mask))
- select {
- case ch := <-w.quit:
- w.quit <- ch
- case w.Events <- event:
- }
- return true
-}
-
-func toWindowsFlags(mask uint64) uint32 {
- var m uint32
- if mask&sysFSACCESS != 0 {
- m |= syscall.FILE_NOTIFY_CHANGE_LAST_ACCESS
- }
- if mask&sysFSMODIFY != 0 {
- m |= syscall.FILE_NOTIFY_CHANGE_LAST_WRITE
- }
- if mask&sysFSATTRIB != 0 {
- m |= syscall.FILE_NOTIFY_CHANGE_ATTRIBUTES
- }
- if mask&(sysFSMOVE|sysFSCREATE|sysFSDELETE) != 0 {
- m |= syscall.FILE_NOTIFY_CHANGE_FILE_NAME | syscall.FILE_NOTIFY_CHANGE_DIR_NAME
- }
- return m
-}
-
-func toFSnotifyFlags(action uint32) uint64 {
- switch action {
- case syscall.FILE_ACTION_ADDED:
- return sysFSCREATE
- case syscall.FILE_ACTION_REMOVED:
- return sysFSDELETE
- case syscall.FILE_ACTION_MODIFIED:
- return sysFSMODIFY
- case syscall.FILE_ACTION_RENAMED_OLD_NAME:
- return sysFSMOVEDFROM
- case syscall.FILE_ACTION_RENAMED_NEW_NAME:
- return sysFSMOVEDTO
- }
- return 0
-}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/build/build_rpm.sh
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/build/build_rpm.sh b/traffic_monitor_golang/build/build_rpm.sh
new file mode 100755
index 0000000..9ec72a3
--- /dev/null
+++ b/traffic_monitor_golang/build/build_rpm.sh
@@ -0,0 +1,58 @@
+#!/bin/bash
+
+#
+#
+# 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.
+#
+
+function importFunctions() {
+ local script=$(readlink -f "$0")
+ local scriptdir=$(dirname "$script")
+ export TM_DIR=$(dirname "$scriptdir")
+ export TC_DIR=$(dirname "$TM_DIR")
+ functions_sh="$TC_DIR/build/functions.sh"
+ if [[ ! -r $functions_sh ]]; then
+ echo "error: can't find $functions_sh"
+ exit 1
+ fi
+ . "$functions_sh"
+}
+
+#----------------------------------------
+function initBuildArea() {
+ echo "Initializing the build area."
+ mkdir -p "$RPMBUILD"/{SPECS,SOURCES,RPMS,SRPMS,BUILD,BUILDROOT} || { echo "Could not create $RPMBUILD: $?"; exit 1; }
+
+ # tar/gzip the source
+ local tm_dest=$(createSourceDir traffic_monitor)
+ cd "$TM_DIR" || \
+ { echo "Could not cd to $TM_DIR: $?"; exit 1; }
+ rsync -av ./ "$tm_dest"/ || \
+ { echo "Could not copy to $tm_dest: $?"; exit 1; }
+ cp "$TM_DIR"/build/*.spec "$RPMBUILD"/SPECS/. || \
+ { echo "Could not copy spec files: $?"; exit 1; }
+
+ cp -r "$TM_DIR"/ "$tm_dest" || { echo "Could not copy $TM_DIR to $tm_dest: $?"; exit 1; }
+
+ tar -czvf "$tm_dest".tgz -C "$RPMBUILD"/SOURCES $(basename $tm_dest) || { echo "Could not create tar archive $tm_dest.tgz: $?"; exit 1; }
+ cp "$TM_DIR"/build/*.spec "$RPMBUILD"/SPECS/. || { echo "Could not copy spec files: $?"; exit 1; }
+
+ echo "The build area has been initialized."
+}
+
+# ---------------------------------------
+
+importFunctions
+checkEnvironment go
+initBuildArea
+buildRpm traffic_monitor
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/build/traffic_monitor.init
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/build/traffic_monitor.init b/traffic_monitor_golang/build/traffic_monitor.init
new file mode 100644
index 0000000..9240ead
--- /dev/null
+++ b/traffic_monitor_golang/build/traffic_monitor.init
@@ -0,0 +1,119 @@
+#!/bin/bash
+
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+# Startup script for traffic_monitor
+#
+#
+# chkconfig: 345 99 10
+# description: traffic_monitor control script
+# processname: traffic_monitor
+
+### BEGIN INIT INFO
+# Provides: traffic_monitor
+# Required-Start: $network $local_fs $syslog
+# Required-Stop: $network $local_fs $syslog
+# Default-Start: 3 4 5
+# Default-Stop: 0 1 2 6
+# Short-Description: start and stop Traffic Monitor
+# Description: Controls all traffic monitor processes at once.
+### END INIT INFO
+
+# Source function library.
+. /etc/init.d/functions
+
+# Source networking configuration.
+. /etc/sysconfig/network
+
+basepath=/opt/traffic_monitor
+binpath=bin
+runpath=var/run
+name=traffic_monitor
+prog=$basepath/$binpath/$name
+lockfile=$basepath/$runpath/$name
+options="--opsCfg /opt/traffic_monitor/conf/traffic_ops.cfg --config /opt/traffic_monitor/conf/traffic_monitor.cfg"
+
+start() {
+ [ "$NETWORKING" = "no" ] && exit 1
+ [ -x $prog ] || exit 5
+
+ #Set file limits
+ # Max open files
+ OPEN_FILE_LIMIT=65536
+ ulimit -n $OPEN_FILE_LIMIT
+ if [ $? -ne 0 ]; then
+ echo -n "Failed to set open file limit to $OPEN_FILE_LIMIT"
+ exit 1
+ fi
+
+ # Start daemons.
+ echo -n $"Starting $name: "
+ daemon nohup $prog $options < /dev/null > /opt/traffic_monitor/var/log/traffic_monitor.log 2>&1 &
+ RETVAL=$?
+ echo
+ [ $RETVAL -eq 0 ] && touch $lockfile
+ return $RETVAL
+}
+
+stop() {
+ echo -n $"Shutting down $name: "
+ killproc $prog
+ RETVAL=$?
+ echo
+ [ $RETVAL -eq 0 ] && rm -f $lockfile
+ return $RETVAL
+}
+
+reload() {
+ echo -n $"Reloading $name: "
+ if [ -n "`pidofproc $prog`" ]; then
+ killproc $prog -HUP
+ else
+ failure $"Reloading $name"
+ fi
+ RETVAL=$?
+ echo
+}
+
+case "$1" in
+ start)
+ start
+ ;;
+ stop)
+ stop
+ ;;
+ status)
+ status $prog
+ ;;
+ restart|force-reload)
+ stop
+ start
+ ;;
+ try-restart|condrestart)
+ if status $prog > /dev/null; then
+ stop
+ start
+ fi
+ ;;
+ reload)
+ reload
+ ;;
+ *)
+ echo $"Usage: $0 {start|stop|status|restart|try-restart|reload|force-reload}"
+ exit 2
+esac
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/build/traffic_monitor.logrotate
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/build/traffic_monitor.logrotate b/traffic_monitor_golang/build/traffic_monitor.logrotate
new file mode 100644
index 0000000..e7073ad
--- /dev/null
+++ b/traffic_monitor_golang/build/traffic_monitor.logrotate
@@ -0,0 +1,26 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+/opt/traffic_monitor/var/log/traffic_monitor.log {
+ compress
+ maxage 30
+ missingok
+ nomail
+ size 10M
+ rotate 5
+ copytruncate
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/build/traffic_monitor.spec
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/build/traffic_monitor.spec b/traffic_monitor_golang/build/traffic_monitor.spec
new file mode 100644
index 0000000..043d977
--- /dev/null
+++ b/traffic_monitor_golang/build/traffic_monitor.spec
@@ -0,0 +1,155 @@
+#
+#
+# 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.
+#
+#
+# RPM spec file for the Go version of Traffic Monitor (tm).
+#
+%define debug_package %{nil}
+Name: traffic_monitor
+Version: %{traffic_control_version}
+Release: %{build_number}
+Summary: Monitor the caches
+Packager: david_neuman2 at Cable dot Comcast dot com
+Vendor: Apache Software Foundation
+Group: Applications/Communications
+License: Apache License, Version 2.0
+URL: https://github.com/apache/incubator-trafficcontrol
+Source: %{_sourcedir}/traffic_monitor-%{traffic_control_version}.tgz
+
+%description
+Installs traffic_monitor
+
+%prep
+
+%setup
+
+%build
+export GOPATH=$(pwd)
+# Create build area with proper gopath structure
+mkdir -p src pkg bin || { echo "Could not create directories in $(pwd): $!"; exit 1; }
+
+go_get_version() {
+ local src=$1
+ local version=$2
+ (
+ cd $src && \
+ git checkout $version && \
+ go get -v \
+ )
+}
+
+# get traffic_ops client
+godir=src/github.com/apache/incubator-trafficcontrol/traffic_ops/client
+( mkdir -p "$godir" && \
+ cd "$godir" && \
+ cp -r "$TC_DIR"/traffic_ops/client/* . && \
+ go get -v \
+) || { echo "Could not build go program at $(pwd): $!"; exit 1; }
+
+#build traffic_monitor binary
+godir=src/github.com/apache/incubator-trafficcontrol/traffic_monitor_golang
+oldpwd=$(pwd)
+( mkdir -p "$godir" && \
+ cd "$godir" && \
+ cp -r "$TC_DIR"/traffic_monitor_golang/* . && \
+ cd traffic_monitor && \
+ go get -d -v && \
+ go build -ldflags "-X main.GitRevision=`git rev-parse HEAD` -X main.BuildTimestamp=`date +'%Y-%M-%dT%H:%M:%s'`" \
+) || { echo "Could not build go program at $(pwd): $!"; exit 1; }
+
+%install
+mkdir -p "${RPM_BUILD_ROOT}"/opt/traffic_monitor
+mkdir -p "${RPM_BUILD_ROOT}"/opt/traffic_monitor/bin
+mkdir -p "${RPM_BUILD_ROOT}"/opt/traffic_monitor/conf
+mkdir -p "${RPM_BUILD_ROOT}"/opt/traffic_monitor/backup
+mkdir -p "${RPM_BUILD_ROOT}"/opt/traffic_monitor/static
+mkdir -p "${RPM_BUILD_ROOT}"/opt/traffic_monitor/var/run
+mkdir -p "${RPM_BUILD_ROOT}"/opt/traffic_monitor/var/log
+mkdir -p "${RPM_BUILD_ROOT}"/etc/init.d
+mkdir -p "${RPM_BUILD_ROOT}"/etc/logrotate.d
+
+src=src/github.com/apache/incubator-trafficcontrol/traffic_monitor_golang
+cp -p "$src"/traffic_monitor/traffic_monitor "${RPM_BUILD_ROOT}"/opt/traffic_monitor/bin/traffic_monitor
+cp "$src"/traffic_monitor/static/index.html "${RPM_BUILD_ROOT}"/opt/traffic_monitor/static/index.html
+cp "$src"/traffic_monitor/static/sorttable.js "${RPM_BUILD_ROOT}"/opt/traffic_monitor/static/sorttable.js
+cp "$src"/conf/traffic_ops.cfg "${RPM_BUILD_ROOT}"/opt/traffic_monitor/conf/traffic_ops.cfg
+cp "$src"/conf/traffic_monitor.cfg "${RPM_BUILD_ROOT}"/opt/traffic_monitor/conf/traffic_monitor.cfg
+cp "$src"/build/traffic_monitor.init "${RPM_BUILD_ROOT}"/etc/init.d/traffic_monitor
+cp "$src"/build/traffic_monitor.logrotate "${RPM_BUILD_ROOT}"/etc/logrotate.d/traffic_monitor
+
+%pre
+/usr/bin/getent group traffic_monitor >/dev/null
+
+if [ $? -ne 0 ]; then
+
+ /usr/sbin/groupadd -g 423 traffic_monitor
+fi
+
+/usr/bin/getent passwd traffic_monitor >/dev/null
+
+if [ $? -ne 0 ]; then
+
+ /usr/sbin/useradd -g traffic_monitor -u 423 -d /opt/traffic_monitor -M traffic_monitor
+
+fi
+
+/usr/bin/passwd -l traffic_monitor >/dev/null
+/usr/bin/chage -E -1 -I -1 -m 0 -M 99999 -W 7 traffic_monitor
+
+if [ -e /etc/init.d/traffic_monitor ]; then
+ /sbin/service traffic_monitor stop
+fi
+
+#don't install over the top of java TM. This is a workaround since yum doesn't respect the Conflicts tag.
+if [[ $(rpm -q traffic_monitor --qf "%{VERSION}-%{RELEASE}") < 1.9.0 ]]
+then
+ echo -e "\n****************\n"
+ echo "A java version of traffic_monitor is installed. Please backup/remove that version before installing the golang version of traffic_monitor."
+ echo -e "\n****************\n"
+ exit 1
+fi
+
+%post
+
+/sbin/chkconfig --add traffic_monitor
+/sbin/chkconfig traffic_monitor on
+
+%files
+%defattr(644, traffic_monitor, traffic_monitor, 755)
+%config(noreplace) /opt/traffic_monitor/conf/traffic_monitor.cfg
+%config(noreplace) /opt/traffic_monitor/conf/traffic_ops.cfg
+%config(noreplace) /etc/logrotate.d/traffic_monitor
+
+%dir /opt/traffic_monitor
+%dir /opt/traffic_monitor/bin
+%dir /opt/traffic_monitor/conf
+%dir /opt/traffic_monitor/backup
+%dir /opt/traffic_monitor/static
+%dir /opt/traffic_monitor/var
+%dir /opt/traffic_monitor/var/log
+%dir /opt/traffic_monitor/var/run
+
+%attr(600, traffic_monitor, traffic_monitor) /opt/traffic_monitor/conf/*
+%attr(600, traffic_monitor, traffic_monitor) /opt/traffic_monitor/static/*
+%attr(755, traffic_monitor, traffic_monitor) /opt/traffic_monitor/bin/*
+%attr(755, traffic_monitor, traffic_monitor) /etc/init.d/traffic_monitor
+
+%preun
+# args for hooks: http://www.ibm.com/developerworks/library/l-rpm2/
+# if $1 = 0, this is an uninstallation, if $1 = 1, this is an upgrade (don't do anything)
+if [ "$1" = "0" ]; then
+ /sbin/chkconfig traffic_monitor off
+ /etc/init.d/traffic_monitor stop
+ /sbin/chkconfig --del traffic_monitor
+fi
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/adapter/adapter.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/adapter/adapter.go b/traffic_monitor_golang/common/adapter/adapter.go
new file mode 100644
index 0000000..bdb3621
--- /dev/null
+++ b/traffic_monitor_golang/common/adapter/adapter.go
@@ -0,0 +1,29 @@
+package adapter
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+
+import (
+ "io"
+)
+
+type Adapter interface {
+ Transform(io.Reader) (interface{}, error)
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/crstates/crstates.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/crstates/crstates.go b/traffic_monitor_golang/common/crstates/crstates.go
new file mode 100644
index 0000000..20e10ca
--- /dev/null
+++ b/traffic_monitor_golang/common/crstates/crstates.go
@@ -0,0 +1,37 @@
+package crstates
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+
+type Cache struct {
+ Name string `json:"name,omitempty"`
+ Available bool `json:"isAvailable"`
+}
+
+type DeliveryService struct {
+ Name string `json:"name,omitempty"`
+ DisabledLocations []string `json:"disabledLocations"`
+ Available bool `json:"isAvailable"`
+}
+
+type CRStates struct {
+ Caches map[string]Cache `json:"caches"`
+ DeliveryServices map[string]DeliveryService `json:"deliveryServices"`
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/fetcher/fetcher.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/fetcher/fetcher.go b/traffic_monitor_golang/common/fetcher/fetcher.go
new file mode 100644
index 0000000..75b4295
--- /dev/null
+++ b/traffic_monitor_golang/common/fetcher/fetcher.go
@@ -0,0 +1,100 @@
+package fetcher
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+import (
+ "fmt"
+ "io/ioutil"
+ "net/http"
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/handler"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/log"
+ "github.com/davecheney/gmx"
+)
+
+type Fetcher interface {
+ Fetch(string, string, uint64, chan<- uint64)
+}
+
+type HttpFetcher struct {
+ Client *http.Client
+ Headers map[string]string
+ Handler handler.Handler
+ Counters
+}
+
+type Result struct {
+ Source string
+ Data []byte
+ Error error
+}
+
+type Counters struct {
+ Success *gmx.Counter
+ Fail *gmx.Counter
+ Pending *gmx.Gauge
+}
+
+func (f HttpFetcher) Fetch(id string, url string, pollId uint64, pollFinishedChan chan<- uint64) {
+ log.Debugf("poll %v %v fetch start\n", pollId, time.Now())
+ req, err := http.NewRequest("GET", url, nil)
+ // TODO: change this to use f.Headers. -jse
+ req.Header.Set("User-Agent", "traffic_monitor/1.0") // TODO change to 2.0?
+ req.Header.Set("Connection", "keep-alive")
+ if f.Pending != nil {
+ f.Pending.Inc()
+ }
+ startReq := time.Now()
+ response, err := f.Client.Do(req)
+ reqTime := time.Now().Sub(startReq)
+ if f.Pending != nil {
+ f.Pending.Dec()
+ }
+ defer func() {
+ if response != nil && response.Body != nil {
+ ioutil.ReadAll(response.Body) // TODO determine if necessary
+ response.Body.Close()
+ }
+ }()
+
+ if err == nil && response == nil {
+ err = fmt.Errorf("err nil and response nil")
+ }
+ if err == nil && response != nil && (response.StatusCode < 200 || response.StatusCode > 299) {
+ err = fmt.Errorf("bad status: %v", response.StatusCode)
+ }
+ if err != nil {
+ err = fmt.Errorf("id %v url %v fetch error: %v", id, url, err)
+ }
+
+ if err == nil && response != nil {
+ if f.Success != nil {
+ f.Success.Inc()
+ }
+ log.Debugf("poll %v %v fetch end\n", pollId, time.Now())
+ f.Handler.Handle(id, response.Body, reqTime, err, pollId, pollFinishedChan)
+ } else {
+ if f.Fail != nil {
+ f.Fail.Inc()
+ }
+ f.Handler.Handle(id, nil, reqTime, err, pollId, pollFinishedChan)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/handler/handler.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/handler/handler.go b/traffic_monitor_golang/common/handler/handler.go
new file mode 100644
index 0000000..e4dd84c
--- /dev/null
+++ b/traffic_monitor_golang/common/handler/handler.go
@@ -0,0 +1,68 @@
+package handler
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+import (
+ "encoding/json"
+ "io"
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/log"
+)
+
+const (
+ NOTIFY_NEVER = iota
+ NOTIFY_CHANGE
+ NOTIFY_ALWAYS
+)
+
+type Handler interface {
+ Handle(string, io.Reader, time.Duration, error, uint64, chan<- uint64)
+}
+
+type OpsConfigFileHandler struct {
+ Content interface{}
+ ResultChannel chan interface{}
+ OpsConfigChannel chan OpsConfig
+}
+
+type OpsConfig struct {
+ Username string `json:"username"`
+ Password string `json:"password"`
+ Url string `json:"url"`
+ Insecure bool `json:"insecure"`
+ CdnName string `json:"cdnName"`
+ HttpListener string `json:"httpListener"`
+}
+
+func (handler OpsConfigFileHandler) Listen() {
+ for {
+ result := <-handler.ResultChannel
+ var toc OpsConfig
+
+ err := json.Unmarshal(result.([]byte), &toc)
+
+ if err != nil {
+ log.Errorf("Could not unmarshal Ops Config JSON: %s\n", err)
+ } else {
+ handler.OpsConfigChannel <- toc
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/instrumentation/instrumentation.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/instrumentation/instrumentation.go b/traffic_monitor_golang/common/instrumentation/instrumentation.go
new file mode 100644
index 0000000..12e29b4
--- /dev/null
+++ b/traffic_monitor_golang/common/instrumentation/instrumentation.go
@@ -0,0 +1,31 @@
+package instrumentation
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+
+import (
+ "github.com/davecheney/gmx"
+)
+
+var TimerFail *gmx.Counter
+
+func init() {
+ TimerFail = gmx.NewCounter("timerFail")
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/log/log.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/log/log.go b/traffic_monitor_golang/common/log/log.go
new file mode 100644
index 0000000..b3b567b
--- /dev/null
+++ b/traffic_monitor_golang/common/log/log.go
@@ -0,0 +1,112 @@
+// Inspired by https://www.goinggo.net/2013/11/using-log-package-in-go.html
+package log
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+import (
+ "fmt"
+ "io"
+ "log"
+ "time"
+)
+
+var (
+ Debug *log.Logger
+ Info *log.Logger
+ Warning *log.Logger
+ Error *log.Logger
+ Event *log.Logger
+)
+
+func Init(eventW, errW, warnW, infoW, debugW io.Writer) {
+ Debug = log.New(debugW, "DEBUG: ", log.Lshortfile)
+ Info = log.New(infoW, "INFO: ", log.Lshortfile)
+ Warning = log.New(warnW, "WARNING: ", log.Lshortfile)
+ Error = log.New(errW, "ERROR: ", log.Lshortfile)
+ Event = log.New(eventW, "", 0)
+}
+
+const timeFormat = time.RFC3339Nano
+const stackFrame = 3
+
+func Errorf(format string, v ...interface{}) {
+ Error.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintf(format, v...))
+}
+func Errorln(v ...interface{}) {
+ Error.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintln(v...))
+}
+func Warnf(format string, v ...interface{}) {
+ Warning.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintf(format, v...))
+}
+func Warnln(v ...interface{}) {
+ Warning.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintln(v...))
+}
+func Infof(format string, v ...interface{}) {
+ Info.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintf(format, v...))
+}
+func Infoln(v ...interface{}) {
+ Info.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintln(v...))
+}
+func Debugf(format string, v ...interface{}) {
+ Debug.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintf(format, v...))
+}
+func Debugln(v ...interface{}) {
+ Debug.Output(stackFrame, time.Now().Format(timeFormat)+": "+fmt.Sprintln(v...))
+}
+
+// event log entries (TM event.log, TR access.log, etc)
+func Eventf(t time.Time, format string, v ...interface{}) {
+ // 1484001185.287 ...
+ Event.Printf("%.3f %s", float64(t.Unix())+(float64(t.Nanosecond())/1e9), fmt.Sprintf(format, v...))
+}
+
+// Close calls `Close()` on the given Closer, and logs any error. On error, the context is logged, followed by a colon, the error message, and a newline. This is primarily designed to be used in `defer`, for example, `defer log.Close(resp.Body, "readData fetching /foo/bar")`.
+func Close(c io.Closer, context string) {
+ err := c.Close()
+ if err != nil {
+ Errorf("%v: %v", context, err)
+ }
+}
+
+// Closef acts like Close, with a given format string and values, followed by a colon, the error message, and a newline. The given values are not coerced, concatenated, or printed unless an error occurs, so this is more efficient than `Close()`.
+func Closef(c io.Closer, contextFormat string, v ...interface{}) {
+ err := c.Close()
+ if err != nil {
+ Errorf(contextFormat, v...)
+ Errorf(": %v", err)
+ }
+}
+
+// Write calls `Write()` on the given Writer, and logs any error. On error, the context is logged, followed by a colon, the error message, and a newline.
+func Write(w io.Writer, b []byte, context string) {
+ _, err := w.Write(b)
+ if err != nil {
+ Errorf("%v: %v", context, err)
+ }
+}
+
+// Writef acts like Write, with a given format string and values, followed by a colon, the error message, and a newline. The given values are not coerced, concatenated, or printed unless an error occurs, so this is more efficient than `Write()`.
+func Writef(w io.Writer, b []byte, contextFormat string, v ...interface{}) {
+ _, err := w.Write(b)
+ if err != nil {
+ Errorf(contextFormat, v...)
+ Errorf(": %v", err)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/poller/heap.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/poller/heap.go b/traffic_monitor_golang/common/poller/heap.go
new file mode 100644
index 0000000..e15d61b
--- /dev/null
+++ b/traffic_monitor_golang/common/poller/heap.go
@@ -0,0 +1,106 @@
+package poller
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+import (
+ "sync"
+ "time"
+)
+
+type HeapPollInfo struct {
+ Info HTTPPollInfo
+ Next time.Time
+}
+
+// Heap implements a Heap from Introduction to Algorithms (Cormen et al). A Heap allows fase access of the maximum object, in this case the latest Next time, and O(log(n)) insert. This Heap is specifically designed to be used as a Priority Queue.
+type Heap struct {
+ m sync.Mutex
+ info []HeapPollInfo
+ PollerID int64
+}
+
+func left(i int) int {
+ return 2*i + 1
+}
+
+func right(i int) int {
+ return 2*i + 2
+}
+
+// TODO benchmark directly replacing this, to see if Go inlines the function call
+func parent(i int) int {
+ return (i - 1) / 2
+}
+
+func (h *Heap) heapify(i int) {
+ l := left(i)
+ r := right(i)
+ var largest int
+ if l < len(h.info) && h.info[i].Next.After(h.info[l].Next) {
+ largest = l
+ } else {
+ largest = i
+ }
+
+ if r < len(h.info) && h.info[largest].Next.After(h.info[r].Next) {
+ largest = r
+ }
+
+ if largest != i {
+ h.info[i], h.info[largest] = h.info[largest], h.info[i]
+ h.heapify(largest)
+ }
+}
+
+func (h *Heap) increaseKey(i int, key HeapPollInfo) {
+ if h.info[i].Next.After(key.Next) {
+ panic("Poll.Heap.increaseKey got key smaller than index")
+ }
+
+ h.info[i] = key
+
+ for i > 0 && h.info[parent(i)].Next.After(h.info[i].Next) {
+ h.info[i], h.info[parent(i)] = h.info[parent(i)], h.info[i]
+ i = parent(i)
+ }
+}
+
+// Pop gets the latest time from the heap. Implements Algorithms HEAP-EXTRACT-MAX.
+// Returns the info with the latest time, and false if the heap is empty.
+func (h *Heap) Pop() (HeapPollInfo, bool) {
+ h.m.Lock()
+ defer h.m.Unlock()
+ if len(h.info) == 0 {
+ return HeapPollInfo{}, false
+ }
+ max := h.info[0]
+ h.info[0] = h.info[len(h.info)-1]
+ h.info = h.info[:len(h.info)-1]
+ h.heapify(0)
+ return max, true
+}
+
+// Pop gets the latest time from the heap. Implements Algorithms MAX-HEAP-INSERT.
+func (h *Heap) Push(key HeapPollInfo) {
+ h.m.Lock()
+ defer h.m.Unlock()
+ h.info = append(h.info, HeapPollInfo{Next: time.Unix(1<<63-1, 0)})
+ h.increaseKey(len(h.info)-1, key)
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/poller/heap_test.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/poller/heap_test.go b/traffic_monitor_golang/common/poller/heap_test.go
new file mode 100644
index 0000000..f943c96
--- /dev/null
+++ b/traffic_monitor_golang/common/poller/heap_test.go
@@ -0,0 +1,162 @@
+package poller
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+import (
+ "fmt"
+ "math/rand"
+ "testing"
+ "time"
+)
+
+func TestHeap(t *testing.T) {
+ h := &Heap{}
+
+ num := 100
+ for i := 0; i < num; i++ {
+ h.Push(HeapPollInfo{
+ Info: HTTPPollInfo{
+ Interval: time.Second * time.Duration(8),
+ ID: fmt.Sprintf("%v", i),
+ },
+ Next: time.Now().Add(time.Second * time.Duration(i)), // time.Duration((i%2)*-1)
+ })
+ }
+
+ for i := 0; i < num; i++ {
+ val, ok := h.Pop()
+ if !ok {
+ t.Errorf("expected pop ID %v got empty heap", i)
+ } else if val.Info.ID != fmt.Sprintf("%v", i) {
+ t.Errorf("expected pop ID %v got %v next %v", i, val.Info.ID, val.Next)
+ }
+ }
+}
+
+func TestHeapRandom(t *testing.T) {
+ h := &Heap{}
+
+ num := 10
+ for i := 0; i < num; i++ {
+ h.Push(HeapPollInfo{
+ Info: HTTPPollInfo{
+ Interval: time.Second * time.Duration(8),
+ ID: fmt.Sprintf("%v", i),
+ },
+ Next: time.Now().Add(time.Duration(rand.Int63())),
+ })
+ }
+
+ previousTime := time.Now()
+ for i := 0; i < num; i++ {
+ val, ok := h.Pop()
+ if !ok {
+ t.Errorf("expected pop ID %v got empty heap", i)
+ } else if previousTime.After(val.Next) {
+ t.Errorf("heap pop %v < previous %v expected >", val.Next, previousTime)
+ }
+ previousTime = val.Next
+ }
+}
+
+func TestHeapRandomPopping(t *testing.T) {
+ h := &Heap{}
+
+ randInfo := func(id int) HeapPollInfo {
+ return HeapPollInfo{
+ Info: HTTPPollInfo{
+ Interval: time.Second * time.Duration(8),
+ ID: fmt.Sprintf("%v", id),
+ },
+ Next: time.Now().Add(time.Duration(rand.Int63())),
+ }
+ }
+
+ num := 10
+ for i := 0; i < num; i++ {
+ h.Push(randInfo(i))
+ }
+
+ previousTime := time.Now()
+ for i := 0; i < num/2; i++ {
+ val, ok := h.Pop()
+ if !ok {
+ t.Errorf("expected pop ID %v got empty heap", i)
+ } else if previousTime.After(val.Next) {
+ t.Errorf("heap pop %v < previous %v expected >", val.Next, previousTime)
+ }
+ previousTime = val.Next
+ }
+
+ for i := 0; i < num; i++ {
+ h.Push(randInfo(i))
+ }
+ val, ok := h.Pop()
+ if !ok {
+ t.Errorf("expected pop, got empty heap")
+ } else {
+ previousTime = val.Next
+ }
+
+ for i := 0; i < num; i++ {
+ val, ok := h.Pop()
+ if !ok {
+ t.Errorf("expected pop ID %v got empty heap", i)
+ } else if previousTime.After(val.Next) {
+ t.Errorf("heap pop %v < previous %v expected >", val.Next, previousTime)
+ }
+ previousTime = val.Next
+ }
+
+ for i := 0; i < num; i++ {
+ h.Push(randInfo(i))
+ }
+ val, ok = h.Pop()
+ if !ok {
+ t.Errorf("expected pop, got empty heap")
+ } else {
+ previousTime = val.Next
+ }
+
+ for i := 0; i < num; i++ {
+ val, ok := h.Pop()
+ if !ok {
+ t.Errorf("expected pop ID %v got empty heap", i)
+ } else if previousTime.After(val.Next) {
+ t.Errorf("heap pop %v < previous %v expected >", val.Next, previousTime)
+ }
+ previousTime = val.Next
+ }
+
+ for i := 0; i < num/2-2; i++ { // -2 for the two we manually popped in order to get the max
+ val, ok := h.Pop()
+ if !ok {
+ t.Errorf("expected pop ID %v got empty heap", i)
+ } else if previousTime.After(val.Next) {
+ t.Errorf("heap pop %v < previous %v expected >", val.Next, previousTime)
+ }
+ previousTime = val.Next
+ }
+
+ val, ok = h.Pop()
+ if ok {
+ t.Errorf("expected empty, got %+v", val)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/poller/poller.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/poller/poller.go b/traffic_monitor_golang/common/poller/poller.go
new file mode 100644
index 0000000..cae41de
--- /dev/null
+++ b/traffic_monitor_golang/common/poller/poller.go
@@ -0,0 +1,399 @@
+package poller
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+import (
+ "io/ioutil"
+ "math/rand"
+ "net/http"
+ "os"
+ "sync/atomic"
+ "time"
+
+ "gopkg.in/fsnotify.v1"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/fetcher"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/handler"
+ instr "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/instrumentation"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/log"
+ towrap "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/trafficopswrapper" // TODO move to common
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
+)
+
+type Poller interface {
+ Poll()
+}
+
+type HttpPoller struct {
+ Config HttpPollerConfig
+ ConfigChannel chan HttpPollerConfig
+ FetcherTemplate fetcher.HttpFetcher // FetcherTemplate has all the constant settings, and is copied to create fetchers with custom HTTP client timeouts.
+ TickChan chan uint64
+}
+
+type PollConfig struct {
+ URL string
+ Timeout time.Duration
+ Handler handler.Handler
+}
+
+type HttpPollerConfig struct {
+ Urls map[string]PollConfig
+ Interval time.Duration
+ // noSleep indicates to use the InsomniacPoller. Note this is only used with the initial Poll call, which decides which Poller mechanism to use. After that, this is ignored when the HttpPollerConfig is passed over the ConfigChannel.
+ noSleep bool
+}
+
+// NewHTTP creates and returns a new HttpPoller.
+// If tick is false, HttpPoller.TickChan() will return nil. If noSleep is true, the poller will busywait instead of sleeping, and use a single goroutine which dispatches polls instead of a goroutine per poll.
+func NewHTTP(
+ interval time.Duration,
+ tick bool,
+ httpClient *http.Client,
+ counters fetcher.Counters,
+ fetchHandler handler.Handler,
+ noSleep bool,
+) HttpPoller {
+ var tickChan chan uint64
+ if tick {
+ tickChan = make(chan uint64)
+ }
+ return HttpPoller{
+ TickChan: tickChan,
+ ConfigChannel: make(chan HttpPollerConfig),
+ Config: HttpPollerConfig{
+ Interval: interval,
+ noSleep: noSleep,
+ },
+ FetcherTemplate: fetcher.HttpFetcher{
+ Handler: fetchHandler,
+ Client: httpClient,
+ Counters: counters,
+ },
+ }
+}
+
+type FilePoller struct {
+ File string
+ ResultChannel chan interface{}
+ NotificationChannel chan int
+}
+
+type MonitorConfigPoller struct {
+ Session towrap.ITrafficOpsSession
+ SessionChannel chan towrap.ITrafficOpsSession
+ ConfigChannel chan to.TrafficMonitorConfigMap
+ OpsConfigChannel chan handler.OpsConfig
+ Interval time.Duration
+ OpsConfig handler.OpsConfig
+}
+
+// Creates and returns a new HttpPoller.
+// If tick is false, HttpPoller.TickChan() will return nil
+func NewMonitorConfig(interval time.Duration) MonitorConfigPoller {
+ return MonitorConfigPoller{
+ Interval: interval,
+ SessionChannel: make(chan towrap.ITrafficOpsSession),
+ ConfigChannel: make(chan to.TrafficMonitorConfigMap),
+ OpsConfigChannel: make(chan handler.OpsConfig),
+ }
+}
+
+func (p MonitorConfigPoller) Poll() {
+ tick := time.NewTicker(p.Interval)
+ defer tick.Stop()
+ for {
+ select {
+ case opsConfig := <-p.OpsConfigChannel:
+ log.Infof("MonitorConfigPoller: received new opsConfig: %v\n", opsConfig)
+ p.OpsConfig = opsConfig
+ case session := <-p.SessionChannel:
+ log.Infof("MonitorConfigPoller: received new session: %v\n", session)
+ p.Session = session
+ case <-tick.C:
+ if p.Session != nil && p.OpsConfig.CdnName != "" {
+ monitorConfig, err := p.Session.TrafficMonitorConfigMap(p.OpsConfig.CdnName)
+
+ if err != nil {
+ log.Errorf("MonitorConfigPoller: %s\n %v\n", err, monitorConfig)
+ } else {
+ log.Debugln("MonitorConfigPoller: fetched monitorConfig")
+ p.ConfigChannel <- *monitorConfig
+ }
+ } else {
+ log.Warnln("MonitorConfigPoller: skipping this iteration, Session is nil")
+ }
+ }
+ }
+}
+
+var debugPollNum uint64
+
+type HTTPPollInfo struct {
+ Interval time.Duration
+ Timeout time.Duration
+ ID string
+ URL string
+ Handler handler.Handler
+}
+
+func (p HttpPoller) Poll() {
+ if p.Config.noSleep {
+ log.Debugf("HttpPoller using InsomniacPoll\n")
+ p.InsomniacPoll()
+ } else {
+ log.Debugf("HttpPoller using SleepPoll\n")
+ p.SleepPoll()
+ }
+}
+
+func (p HttpPoller) SleepPoll() {
+ // iterationCount := uint64(0)
+ // iterationCount++ // on tick<:
+ // case p.TickChan <- iterationCount:
+ killChans := map[string]chan<- struct{}{}
+ for newConfig := range p.ConfigChannel {
+ deletions, additions := diffConfigs(p.Config, newConfig)
+ for _, id := range deletions {
+ killChan := killChans[id]
+ go func() { killChan <- struct{}{} }() // go - we don't want to wait for old polls to die.
+ delete(killChans, id)
+ }
+ for _, info := range additions {
+ kill := make(chan struct{})
+ killChans[info.ID] = kill
+
+ fetcher := p.FetcherTemplate
+ if info.Timeout != 0 { // if the timeout isn't explicitly set, use the template value.
+ c := *fetcher.Client
+ fetcher.Client = &c // copy the client, so we don't change other fetchers.
+ fetcher.Client.Timeout = info.Timeout
+ }
+ go sleepPoller(info.Interval, info.ID, info.URL, fetcher, kill)
+ }
+ p.Config = newConfig
+ }
+}
+
+func mustDie(die <-chan struct{}) bool {
+ select {
+ case <-die:
+ return true
+ default:
+ }
+ return false
+}
+
+// TODO iterationCount and/or p.TickChan?
+func sleepPoller(interval time.Duration, id string, url string, fetcher fetcher.Fetcher, die <-chan struct{}) {
+ pollSpread := time.Duration(rand.Float64()*float64(interval/time.Nanosecond)) * time.Nanosecond
+ time.Sleep(pollSpread)
+ tick := time.NewTicker(interval)
+ lastTime := time.Now()
+ for {
+ select {
+ case <-tick.C:
+ realInterval := time.Now().Sub(lastTime)
+ if realInterval > interval+(time.Millisecond*100) {
+ instr.TimerFail.Inc()
+ log.Debugf("Intended Duration: %v Actual Duration: %v\n", interval, realInterval)
+ }
+ lastTime = time.Now()
+
+ pollId := atomic.AddUint64(&debugPollNum, 1)
+ pollFinishedChan := make(chan uint64)
+ log.Debugf("poll %v %v start\n", pollId, time.Now())
+ go fetcher.Fetch(id, url, pollId, pollFinishedChan) // TODO persist fetcher, with its own die chan?
+ <-pollFinishedChan
+ case <-die:
+ tick.Stop()
+ return
+ }
+ }
+}
+
+const InsomniacPollerEmptySleepDuration = time.Millisecond * time.Duration(100)
+
+// InsomniacPoll polls using a single thread, which never sleeps. This exists to work around a bug observed in OpenStack CentOS 6.5 kernel 2.6.32 wherin sleep gets progressively slower. This should be removed and Poll() changed to call SleepPoll() when the bug is tracked down and fixed for production.
+func (p HttpPoller) InsomniacPoll() {
+ // iterationCount := uint64(0)
+ // iterationCount++ // on tick<:
+ // case p.TickChan <- iterationCount:
+ killChan := make(chan struct{})
+ pollRunning := false // TODO find less awkward way to not kill the first loop
+ pollerId := rand.Int63()
+ for newCfg := range p.ConfigChannel {
+ // TODO add a more efficient function than diffConfigs for this func, since we only need to know whether anything changed
+ deletions, additions := diffConfigs(p.Config, newCfg)
+ if len(deletions) == 0 && len(additions) == 0 {
+ continue
+ }
+
+ if pollRunning {
+ killChan <- struct{}{}
+ }
+ pollRunning = true
+
+ polls := []HTTPPollInfo{}
+ for id, pollCfg := range newCfg.Urls {
+ polls = append(polls, HTTPPollInfo{
+ Interval: newCfg.Interval - InsomniacPollerEmptySleepDuration,
+ ID: id,
+ URL: pollCfg.URL,
+ Timeout: pollCfg.Timeout,
+ })
+ }
+ go insomniacPoller(pollerId, polls, p.FetcherTemplate, killChan)
+ p.Config = newCfg
+ }
+}
+
+func insomniacPoller(pollerId int64, polls []HTTPPollInfo, fetcherTemplate fetcher.HttpFetcher, die <-chan struct{}) {
+ heap := Heap{PollerID: pollerId}
+ start := time.Now()
+ fetchers := map[string]fetcher.Fetcher{}
+ for _, p := range polls {
+ spread := time.Duration(rand.Float64()*float64(p.Interval/time.Nanosecond)) * time.Nanosecond
+ heap.Push(HeapPollInfo{Info: p, Next: start.Add(spread)})
+
+ fetcher := fetcherTemplate
+ if p.Timeout != 0 { // if the timeout isn't explicitly set, use the template value.
+ c := *fetcher.Client
+ fetcher.Client = &c // copy the client, so we don't change other fetchers.
+ fetcher.Client.Timeout = p.Timeout
+ }
+ fetchers[p.ID] = fetcher
+ }
+
+ timeMax := func(a time.Time, b time.Time) time.Time {
+ if a.After(b) {
+ return a
+ }
+ return b
+ }
+
+ poll := func(p HeapPollInfo) {
+ start := time.Now()
+ pollId := atomic.AddUint64(&debugPollNum, 1)
+ // TODO change pollFinishedChan to callback, for performance
+ pollFinishedChan := make(chan uint64)
+
+ go fetchers[p.Info.ID].Fetch(p.Info.ID, p.Info.URL, pollId, pollFinishedChan) // TODO persist fetcher, with its own die chan?
+ <-pollFinishedChan
+ now := time.Now()
+ p.Next = timeMax(start.Add(p.Info.Interval), now)
+ heap.Push(p)
+ }
+
+ for {
+ p, ok := heap.Pop()
+ if !ok {
+ ThreadSleep(InsomniacPollerEmptySleepDuration)
+ continue
+ }
+ if mustDie(die) {
+ return
+ }
+ ThreadSleep(p.Next.Sub(time.Now()))
+ go poll(p)
+ }
+}
+
+func (p FilePoller) Poll() {
+ // initial read before watching for changes
+ contents, err := ioutil.ReadFile(p.File)
+
+ if err != nil {
+ log.Errorf("reading %s: %s\n", p.File, err)
+ os.Exit(1) // TODO: this is a little drastic -jse
+ } else {
+ p.ResultChannel <- contents
+ }
+
+ watcher, _ := fsnotify.NewWatcher()
+ watcher.Add(p.File)
+
+ for {
+ select {
+ case event := <-watcher.Events:
+ if event.Op&fsnotify.Write == fsnotify.Write {
+ contents, err := ioutil.ReadFile(p.File)
+
+ if err != nil {
+ log.Errorf("opening %s: %s\n", p.File, err)
+ } else {
+ p.ResultChannel <- contents
+ }
+ }
+ case err := <-watcher.Errors:
+ log.Errorln(time.Now(), "error:", err)
+ }
+ }
+}
+
+// diffConfigs takes the old and new configs, and returns a list of deleted IDs, and a list of new polls to do
+func diffConfigs(old HttpPollerConfig, new HttpPollerConfig) ([]string, []HTTPPollInfo) {
+ deletions := []string{}
+ additions := []HTTPPollInfo{}
+
+ if old.Interval != new.Interval {
+ for id, _ := range old.Urls {
+ deletions = append(deletions, id)
+ }
+ for id, pollCfg := range new.Urls {
+ additions = append(additions, HTTPPollInfo{
+ Interval: new.Interval,
+ ID: id,
+ URL: pollCfg.URL,
+ Timeout: pollCfg.Timeout,
+ })
+ }
+ return deletions, additions
+ }
+
+ for id, oldPollCfg := range old.Urls {
+ newPollCfg, newIdExists := new.Urls[id]
+ if !newIdExists {
+ deletions = append(deletions, id)
+ } else if newPollCfg != oldPollCfg {
+ deletions = append(deletions, id)
+ additions = append(additions, HTTPPollInfo{
+ Interval: new.Interval,
+ ID: id,
+ URL: newPollCfg.URL,
+ Timeout: newPollCfg.Timeout,
+ })
+ }
+ }
+
+ for id, newPollCfg := range new.Urls {
+ _, oldIdExists := old.Urls[id]
+ if !oldIdExists {
+ additions = append(additions, HTTPPollInfo{
+ Interval: new.Interval,
+ ID: id,
+ URL: newPollCfg.URL,
+ Timeout: newPollCfg.Timeout,
+ })
+ }
+ }
+
+ return deletions, additions
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/poller/threadsleep.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/poller/threadsleep.go b/traffic_monitor_golang/common/poller/threadsleep.go
new file mode 100644
index 0000000..d4a67cc
--- /dev/null
+++ b/traffic_monitor_golang/common/poller/threadsleep.go
@@ -0,0 +1,36 @@
+// +build !linux
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 poller
+
+import (
+ "runtime"
+ "time"
+)
+
+// ThreadSleep actually busywaits for the given duration. This is becuase Go doesn't have Mac and Windows nanosleep syscalls, and `Sleep` sleeps for progressively longer than requested.
+func ThreadSleep(d time.Duration) {
+ // TODO fix to not busywait on Mac, Windows. We can't simply Sleep, because Sleep gets progressively slower as the app runs, due to a Go runtime issue. If this is changed, you MUST verify the poll doesn't get slower after the app runs for several days.
+ end := time.Now().Add(d)
+ for end.After(time.Now()) {
+ runtime.Gosched()
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/poller/threadsleep_linux.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/poller/threadsleep_linux.go b/traffic_monitor_golang/common/poller/threadsleep_linux.go
new file mode 100644
index 0000000..ba7c379
--- /dev/null
+++ b/traffic_monitor_golang/common/poller/threadsleep_linux.go
@@ -0,0 +1,42 @@
+// +build linux
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 poller
+
+import (
+ "errors"
+ "golang.org/x/sys/unix"
+ "time"
+)
+
+// ThreadSleep sleeps using the POSIX syscall `nanosleep`. Note this does not sleep the goroutine, but the operating system thread itself. This should only be called by a goroutine which has previously called `LockOSThread`. This exists due to a bug with `time.Sleep` getting progressively slower as the app runs, and should be removed if the bug in Go is fixed.
+func ThreadSleep(d time.Duration) {
+ if d < 0 {
+ d = 0
+ }
+ t := unix.Timespec{}
+ leftover := unix.NsecToTimespec(d.Nanoseconds())
+ err := errors.New("")
+ for err != nil && (leftover.Sec != 0 || leftover.Nsec != 0) {
+ t = leftover
+ err = unix.Nanosleep(&t, &leftover)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/util/join.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/util/join.go b/traffic_monitor_golang/common/util/join.go
new file mode 100644
index 0000000..d5b647b
--- /dev/null
+++ b/traffic_monitor_golang/common/util/join.go
@@ -0,0 +1,60 @@
+package util
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+import (
+ "fmt"
+)
+
+func JoinErrorsString(errs []error) string {
+ joined := JoinErrors(errs)
+
+ if joined == nil {
+ return ""
+ }
+
+ return joined.Error()
+}
+
+func JoinErrors(errs []error) error {
+ return JoinErrorsSep(errs, "")
+}
+
+func JoinErrorsSep(errs []error, separator string) error {
+ if separator == "" {
+ separator = ", "
+ }
+
+ joinedErrors := ""
+
+ for _, err := range errs {
+ if err != nil {
+ joinedErrors += err.Error() + separator
+ }
+ }
+
+ if len(joinedErrors) == 0 {
+ return nil
+ }
+
+ joinedErrors = joinedErrors[:len(joinedErrors)-len(separator)] // strip trailing separator
+
+ return fmt.Errorf("%s", joinedErrors)
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/common/util/num.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/common/util/num.go b/traffic_monitor_golang/common/util/num.go
new file mode 100644
index 0000000..4db2cf9
--- /dev/null
+++ b/traffic_monitor_golang/common/util/num.go
@@ -0,0 +1,35 @@
+package util
+
+// ToNumeric returns a float for any numeric type, and false if the interface does not hold a numeric type.
+// This allows converting unknown numeric types (for example, from JSON) in a single line
+// TODO try to parse string stats as numbers?
+func ToNumeric(v interface{}) (float64, bool) {
+ switch i := v.(type) {
+ case uint8:
+ return float64(i), true
+ case uint16:
+ return float64(i), true
+ case uint32:
+ return float64(i), true
+ case uint64:
+ return float64(i), true
+ case int8:
+ return float64(i), true
+ case int16:
+ return float64(i), true
+ case int32:
+ return float64(i), true
+ case int64:
+ return float64(i), true
+ case float32:
+ return float64(i), true
+ case float64:
+ return i, true
+ case int:
+ return float64(i), true
+ case uint:
+ return float64(i), true
+ default:
+ return 0.0, false
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/conf/traffic_monitor.cfg
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/conf/traffic_monitor.cfg b/traffic_monitor_golang/conf/traffic_monitor.cfg
new file mode 100644
index 0000000..f6698d1
--- /dev/null
+++ b/traffic_monitor_golang/conf/traffic_monitor.cfg
@@ -0,0 +1,22 @@
+{
+ "cache_health_polling_interval_ms": 6000,
+ "cache_stat_polling_interval_ms": 6000,
+ "monitor_config_polling_interval_ms": 5000,
+ "http_timeout_ms": 2000,
+ "peer_polling_interval_ms": 5000,
+ "peer_optimistic": true,
+ "max_events": 200,
+ "max_stat_history": 5,
+ "max_health_history": 5,
+ "health_flush_interval_ms": 20,
+ "stat_flush_interval_ms": 20,
+ "log_location_event": "null",
+ "log_location_error": "stderr",
+ "log_location_warning": "stdout",
+ "log_location_info": "null",
+ "log_location_debug": "null",
+ "serve_read_timeout_ms": 10000,
+ "serve_write_timeout_ms": 10000,
+ "http_poll_no_sleep": false,
+ "static_file_dir": "/opt/traffic_monitor/static/"
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/conf/traffic_ops.cfg
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/conf/traffic_ops.cfg b/traffic_monitor_golang/conf/traffic_ops.cfg
new file mode 100644
index 0000000..b5fcf83
--- /dev/null
+++ b/traffic_monitor_golang/conf/traffic_ops.cfg
@@ -0,0 +1,8 @@
+{
+ "username": "",
+ "password": "",
+ "url": "https://traffic-ops.example.net",
+ "insecure": true,
+ "cdnName": "cdn",
+ "httpListener": ":8080"
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/traffic_monitor/README.md
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/README.md b/traffic_monitor_golang/traffic_monitor/README.md
new file mode 100644
index 0000000..c3534fb
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/README.md
@@ -0,0 +1 @@
+To run: `./build.sh && traffic_monitor --opsCfg ./traffic_ops.cfg -config ./traffic_monitor-example-config.json`
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/traffic_monitor/build.sh
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/build.sh b/traffic_monitor_golang/traffic_monitor/build.sh
new file mode 100755
index 0000000..baf07cc
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/build.sh
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+#!/usr/bin/env bash
+go build -ldflags "-X main.GitRevision=`git rev-parse HEAD` -X main.BuildTimestamp=`date +'%Y-%M-%dT%H:%M:%S'`"
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/traffic_monitor/cache/astats.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/cache/astats.go b/traffic_monitor_golang/traffic_monitor/cache/astats.go
new file mode 100644
index 0000000..96258c8
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/cache/astats.go
@@ -0,0 +1,51 @@
+package cache
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+
+import (
+ "encoding/json"
+)
+
+// Astats contains ATS data returned from the Astats ATS plugin. This includes generic stats, as well as fixed system stats.
+type Astats struct {
+ Ats map[string]interface{} `json:"ats"`
+ System AstatsSystem `json:"system"`
+}
+
+// AstatsSystem represents fixed system stats returne from ATS by the Astats plugin.
+type AstatsSystem struct {
+ InfName string `json:"inf.name"`
+ InfSpeed int `json:"inf.speed"`
+ ProcNetDev string `json:"proc.net.dev"`
+ ProcLoadavg string `json:"proc.loadavg"`
+ ConfigLoadRequest int `json:"configReloadRequests"`
+ LastReloadRequest int `json:"lastReloadRequest"`
+ ConfigReloads int `json:"configReloads"`
+ LastReload int `json:"lastReload"`
+ AstatsLoad int `json:"astatsLoad"`
+}
+
+// Unmarshal unmarshalls the given bytes, which must be JSON Astats data, into an Astats object.
+func Unmarshal(body []byte) (Astats, error) {
+ var aStats Astats
+ err := json.Unmarshal(body, &aStats)
+ return aStats, err
+}