implement wrapper for driver, sets up fifo file for log streaming

driver.go

@@ -1,9 +1,16 @@
 package main
 
 import (
+  "context"
+  "fmt"
+  "io"
+  "net/http"
   "sync"
+  "syscall"
 
   "github.com/docker/docker/daemon/logger"
+  "github.com/pkg/errors"
+  "github.com/tonistiigi/fifo"
 )
 
 type SumoDriver interface {
@@ -17,6 +24,9 @@ type sumoDriver struct {
 }
 
 type sumoLogger struct {
+  httpSourceUrl string
+  client *http.Client
+  fifoLogStream io.ReadWriteCloser
 }
 
 func NewSumoDriver() *sumoDriver {
@@ -26,9 +36,44 @@ func NewSumoDriver() *sumoDriver {
 }
 
 func (sumoDriver *sumoDriver) StartLogging(file string, info logger.Info) error {
+  sumoDriver.mu.Lock()
+  if _, exists := sumoDriver.loggers[file]; exists {
+    sumoDriver.mu.Unlock()
+    return fmt.Errorf("a logger for %q already exists", file)
+  }
+  sumoDriver.mu.Unlock()
+
+  fifoLogStream, err := fifo.OpenFifo(context.Background(), file, syscall.O_RDONLY, 0700)
+	if err != nil {
+		return errors.Wrapf(err, "error opening logger fifo: %q", file)
+	}
+
+  newSumoLogger := &sumoLogger{
+    httpSourceUrl: info.Config[logOptUrl],
+    client: &http.Client{},
+    fifoLogStream: fifoLogStream,
+  }
+
+  sumoDriver.mu.Lock()
+  sumoDriver.loggers[file] = newSumoLogger
+  sumoDriver.mu.Unlock()
+
+  go consumeLogs(newSumoLogger)
+
   return nil
 }
 
 func (sumoDriver *sumoDriver) StopLogging(file string) error {
+	sumoDriver.mu.Lock()
+	logger, exists := sumoDriver.loggers[file]
+	if exists {
+		logger.fifoLogStream.Close()
+		delete(sumoDriver.loggers, file)
+	}
+	sumoDriver.mu.Unlock()
   return nil
 }
+
+func consumeLogs(sumoLogger *sumoLogger) {
+  return
+}

driver_test.go

@@ -0,0 +1,94 @@
+package main
+
+import (
+  "context"
+  "strconv"
+  "testing"
+
+  "github.com/docker/docker/daemon/logger"
+  "github.com/stretchr/testify/assert"
+  "github.com/tonistiigi/fifo"
+  "golang.org/x/sys/unix"
+)
+
+const (
+  filePath  = "/tmp/test"
+  filePath1 = "/tmp/test1"
+  filePath2 = "/tmp/test2"
+)
+
+func TestDrivers (t *testing.T) {
+  testLoggersCount := 100
+
+  for i := 0; i < testLoggersCount; i++ {
+    testFifo, err := fifo.OpenFifo(context.Background(), filePath + strconv.Itoa(i + 1), unix.O_WRONLY|unix.O_CREAT|unix.O_NONBLOCK, 0700)
+    assert.Nil(t, err)
+    defer testFifo.Close()
+  }
+
+  info := logger.Info{
+    Config: map[string]string{
+      logOptUrl: "https://example.org",
+    },
+    ContainerID: "containeriid",
+  }
+
+  t.Run("StartLogging", func(t *testing.T) {
+    testSumoDriver := NewSumoDriver()
+    assert.Equal(t, 0, len(testSumoDriver.loggers), "there should be no loggers when the driver is initialized")
+
+    err := testSumoDriver.StartLogging(filePath1, info)
+    assert.Nil(t, err)
+    assert.Equal(t, 1, len(testSumoDriver.loggers), "there should be one logger after calling StartLogging on driver")
+
+    err = testSumoDriver.StartLogging(filePath1, info)
+    assert.Error(t, err, "trying to call StartLogging for filepath that already exists should return error")
+    assert.Equal(t, 1, len(testSumoDriver.loggers),
+      "there should still be one logger after calling StartLogging for filepath that already exists")
+
+    err = testSumoDriver.StartLogging(filePath2, info)
+    assert.Nil(t, err)
+    assert.Equal(t, 2, len(testSumoDriver.loggers),
+      "there should be two loggers now after calling StartLogging on driver for different filepaths")
+  })
+
+  t.Run("StopLogging", func(t *testing.T) {
+    testSumoDriver := NewSumoDriver()
+    assert.Equal(t, 0, len(testSumoDriver.loggers), "there should be no loggers when the driver is initialized")
+
+    err := testSumoDriver.StopLogging(filePath1)
+    assert.Nil(t, err, "trying to call StopLogging for nonexistent logger should NOT return error")
+    assert.Equal(t, 0, len(testSumoDriver.loggers), "no loggers should be changed after calling StopLogging for nonexistent logger")
+
+    err = testSumoDriver.StartLogging(filePath1, info)
+    assert.Nil(t, err)
+    assert.Equal(t, 1, len(testSumoDriver.loggers), "there should be one logger after calling StartLogging on driver")
+
+    err = testSumoDriver.StopLogging(filePath2)
+    assert.Nil(t, err, "trying to call StopLogging for nonexistent logger should NOT return error")
+    assert.Equal(t, 1, len(testSumoDriver.loggers), "no loggers should be changed after calling StopLogging for nonexistent logger")
+
+    err = testSumoDriver.StopLogging(filePath1)
+    assert.Nil(t, err, "trying to call StopLogging for existing logger should not return error")
+    assert.Equal(t, 0, len(testSumoDriver.loggers), "calling StopLogging on existing logger should remove the logger")
+  })
+
+  t.Run("StartLogging, concurrently", func(t *testing.T) {
+    testSumoDriver := NewSumoDriver()
+    assert.Equal(t, 0, len(testSumoDriver.loggers), "there should be no loggers when the driver is initialized")
+
+    waitForAllLoggers := make(chan int)
+    for i := 0; i < testLoggersCount; i++ {
+      go func(i int) {
+        err := testSumoDriver.StartLogging(filePath + strconv.Itoa(i + 1), info)
+        assert.Nil(t, err)
+        waitForAllLoggers <- i
+      }(i)
+    }
+    for i := 0; i < testLoggersCount; i ++ {
+      <-waitForAllLoggers
+    }
+    assert.Equal(t, testLoggersCount, len(testSumoDriver.loggers),
+      "there should be %v loggers now after calling StartLogging on driver that many times on different filepaths", testLoggersCount)
+  })
+}

main_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-  fileString = "/run/docker/logging/b227f9375aee6474f9195fc53710159e7afbc2af872d6287380dfbffb0a2e224"
+  filePathRequestField = "/tmp/test"
 )
 
 type mockSumoDriver struct {
@@ -49,7 +49,7 @@ func TestHandlers(t *testing.T) {
   t.Run("make StartLogging request with missing ContainerID", func(t *testing.T) {
     defer resetCallsCount(mockSumoDriver)
     req := StartLoggingRequest{
-      File: fileString,
+      File: filePathRequestField,
       Info: logger.Info{
         Config: map[string]string{
           logOptUrl: "https://example.org",
@@ -70,7 +70,7 @@ func TestHandlers(t *testing.T) {
   t.Run(fmt.Sprintf("make StartLogging request with missing log-opt: %s", logOptUrl), func(t *testing.T) {
     defer resetCallsCount(mockSumoDriver)
     req := StartLoggingRequest{
-      File: fileString,
+      File: filePathRequestField,
       Info: logger.Info{
         Config: map[string]string{},
         ContainerID: "containeriid",
@@ -90,7 +90,7 @@ func TestHandlers(t *testing.T) {
   t.Run("make StartLogging request with correct config", func(t *testing.T) {
     defer resetCallsCount(mockSumoDriver)
     req := StartLoggingRequest{
-      File: fileString,
+      File: filePathRequestField,
       Info: logger.Info{
         Config: map[string]string{
           logOptUrl: "https://example.org",
@@ -112,7 +112,7 @@ func TestHandlers(t *testing.T) {
   t.Run("make StopLogging request", func(t *testing.T) {
     defer resetCallsCount(mockSumoDriver)
     req := StopLoggingRequest{
-      File: fileString,
+      File: filePathRequestField,
     }
     resp, respBody, err := makeRequest(stopLoggingPath, req, mockServer)
     if err != nil {
@@ -123,7 +123,6 @@ func TestHandlers(t *testing.T) {
     assert.Equal(t, 1, mockSumoDriver.StopLoggingCallsCount, "should have called StopLogging on the driver exactly once")
     assert.Equal(t, "", respBody.Err, "error message should be empty")
   })
-
 }
 
 func resetCallsCount(m *mockSumoDriver) {