Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
The event socket now integrated into tcp-info (#107)
The event socket now integrated into tcp-info with a basic client.
- Loading branch information
Showing
8 changed files
with
244 additions
and
54 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| package eventsocket | ||
|
|
||
| import ( | ||
| "bufio" | ||
| "context" | ||
| "encoding/json" | ||
| "log" | ||
| "net" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/m-lab/go/rtx" | ||
| "github.com/m-lab/tcp-info/inetdiag" | ||
| ) | ||
|
|
||
| // Handler is the interface that all interested users of the event socket | ||
| // notifications should implement. It has two methods, one called on Open events | ||
| // and one called on Close events. | ||
| type Handler interface { | ||
| Open(timestamp time.Time, uuid string, ID *inetdiag.SockID) | ||
| Close(timestamp time.Time, uuid string) | ||
| } | ||
|
|
||
| // MustRun will read from the passed-in socket filename until the context is | ||
| // cancelled. Any errors are fatal. | ||
| func MustRun(ctx context.Context, socket string, handler Handler) { | ||
| ctx, cancel := context.WithCancel(ctx) | ||
| defer cancel() | ||
| c, err := net.Dial("unix", socket) | ||
| rtx.Must(err, "Could not connect to %q", socket) | ||
| go func() { | ||
| // Close the connection when the context is done. Closing the underlying | ||
| // connection means that the scanner will soon terminate. | ||
| <-ctx.Done() | ||
| c.Close() | ||
| }() | ||
|
|
||
| // By default bufio.Scanner is based on newlines, which is perfect for our JSONL protocol. | ||
| s := bufio.NewScanner(c) | ||
| for s.Scan() { | ||
| var event FlowEvent | ||
| rtx.Must(json.Unmarshal(s.Bytes(), &event), "Could not unmarshall") | ||
| switch event.Event { | ||
| case Open: | ||
| handler.Open(event.Timestamp, event.UUID, event.ID) | ||
| case Close: | ||
| handler.Close(event.Timestamp, event.UUID) | ||
| default: | ||
| log.Println("Unknown event type:", event.Event) | ||
| } | ||
| } | ||
|
|
||
| // s.Err() is supposed to be nil under normal conditions. Scanner objects | ||
| // hide the expected EOF error and return nil after they encounter it, | ||
| // because EOF is the expected error. However, reading on a closed socket | ||
| // doesn't give you an EOF error and the error it does give you is | ||
| // unexported. The error it gives you should be treated the same as EOF, | ||
| // because it corresponds to the connection terminating under normal | ||
| // conditions. Because Scanner hides the EOF error, it should also hide the | ||
| // unexported one. Because Scanner doesn't, we do so here. Other errors | ||
| // should not be hidden. | ||
| err = s.Err() | ||
| if err != nil && strings.Contains(err.Error(), "use of closed network connection") { | ||
| err = nil | ||
| } | ||
| rtx.Must(err, "Scanning of %f died with non-EOF error", socket) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| package eventsocket | ||
|
|
||
| import ( | ||
| "context" | ||
| "io/ioutil" | ||
| "os" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/m-lab/go/rtx" | ||
| "github.com/m-lab/tcp-info/inetdiag" | ||
| ) | ||
|
|
||
| type testHandler struct { | ||
| opens, closes int | ||
| wg sync.WaitGroup | ||
| } | ||
|
|
||
| func (t *testHandler) Open(timestamp time.Time, uuid string, id *inetdiag.SockID) { | ||
| t.opens++ | ||
| t.wg.Done() | ||
| } | ||
|
|
||
| func (t *testHandler) Close(timestamp time.Time, uuid string) { | ||
| t.closes++ | ||
| t.wg.Done() | ||
| } | ||
|
|
||
| func TestClient(t *testing.T) { | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
| dir, err := ioutil.TempDir("", "TestEventSocketClient") | ||
| rtx.Must(err, "Could not create tempdir") | ||
| defer os.RemoveAll(dir) | ||
|
|
||
| srv := New(dir + "/tcpevents.sock").(*server) | ||
| srv.Listen() | ||
| srvCtx, srvCancel := context.WithCancel(context.Background()) | ||
| go srv.Serve(srvCtx) | ||
| defer srvCancel() | ||
|
|
||
| th := &testHandler{} | ||
| clientWg := sync.WaitGroup{} | ||
| clientWg.Add(1) | ||
| go func() { | ||
| MustRun(ctx, dir+"/tcpevents.sock", th) | ||
| clientWg.Done() | ||
| }() | ||
| th.wg.Add(2) | ||
|
|
||
| // Send an open event | ||
| srv.FlowCreated(time.Now(), "fakeuuid", inetdiag.SockID{}) | ||
| // Send a bad event and make sure nothing crashes. | ||
| srv.eventC <- &FlowEvent{ | ||
| Event: TCPEvent(1000), | ||
| Timestamp: time.Now(), | ||
| UUID: "fakeuuid", | ||
| } | ||
| // Send a deletion event | ||
| srv.FlowDeleted(time.Now(), "fakeuuid") | ||
| th.wg.Wait() // Wait until the handler gets two events! | ||
|
|
||
| // Cancel the context and wait until the client stops running. | ||
| cancel() | ||
| clientWg.Wait() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.