1298

Committed 12 May 2020 06:11PM UTC coverage: 79.874%. First build

Build # 1298

Build Type

Pull #289

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/m-lab/ndt-server/commit/<a class=hub.com/m-lab/ndt-server/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/ndt-server/commit/<a class="double-link" href="https://git"><a class=hub.com/m-lab/ndt-server/commit/c2cfb9fae1fb7be8bf9536bf5c27b2c51871575f">c2cfb9fae<a href="https://github.com/m-lab/ndt-server/commit/c2cfb9fae1fb7be8bf9536bf5c27b2c51871575f">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/m-lab/ndt-server/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/m-lab/ndt-server/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/m-lab/ndt-server/commit/c2cfb9fae1fb7be8bf9536bf5c27b2c51871575f&quot;&gt;c2cfb9fae&lt;/a&gt;&lt;a href=&quot;https://github.com/m-lab/ndt-server/commit/c2cfb9fae1fb7be8bf9536bf5c27b2c51871575f&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/m-lab/ndt-server/commit/c2cfb9fae1fb7be8bf9536bf5c27b2c5&lt;/a&gt;1871575f">&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/m-lab/ndt-server/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/m-lab/ndt-server/commit/<a class="double-link" href="https://github.com/m-lab/ndt-server/commit/44b5e46c14eb3251b094d893608556eece61f645">44b5e46c1</a><a href="https://github.com/m-lab/ndt-server/commit/c2cfb9fae1fb7be8bf9536bf5c27b2c51871575f">&amp;quot;&amp;gt;44b5e46c1&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;44b5e46c1&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/m-lab/ndt-server/commit/c2cfb9fae1fb7be8bf9536bf5c27b2c518715&lt;/a&gt;75f&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/m-lab/ndt-server/commit/<a class="double-link" href="https://github.com/m-lab/ndt-server/commit/7b33611c2">7b33611c2">7b33611c2</a>

Pull Request Pull Request #289: Add metrics shared with clients to ArchivalData

Run Details

11 of 11 new or added lines in 2 files covered. (100.0%)

1651 of 2067 relevant lines covered (79.87%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.21

/ndt5/c2s/c2s.go

package c2s

import (
        "context"
        "log"
        "strconv"
        "time"

        "github.com/m-lab/go/warnonerror"
        "github.com/m-lab/ndt-server/ndt5/metrics"
        "github.com/m-lab/ndt-server/ndt5/ndt"
        "github.com/m-lab/ndt-server/ndt5/protocol"
        "github.com/m-lab/ndt-server/ndt5/web100"
)

// ArchivalData is the data saved by the C2S test. If a researcher wants deeper
// data, then they should use the UUID to get deeper data from tcp-info.
type ArchivalData struct {
        // The server and client IP are here as well as in the containing struct
        // because happy eyeballs means that we may have a IPv4 control connection
        // causing a IPv6 connection to the test port or vice versa.
        ServerIP   string
        ServerPort int
        ClientIP   string
        ClientPort int

        // This is the only field that is really required.
        UUID string

        // These fields are here to enable analyses that don't require joining with tcp-info data.
        StartTime          time.Time
        EndTime            time.Time
        MeanThroughputMbps float64
        // TODO: Add TCPEngine (bbr, cubic, reno, etc.)

        Error string `json:",omitempty"`
}

// ManageTest manages the c2s test lifecycle.
func ManageTest(ctx context.Context, controlConn protocol.Connection, s ndt.Server) (record *ArchivalData, err error) {
        localContext, localCancel := context.WithTimeout(ctx, 30*time.Second)
        defer localCancel()
        defer func() {
                if err != nil && record != nil {
                        record.Error = err.Error()
                }
        }()
        record = &ArchivalData{}

        m := controlConn.Messager()
        connType := s.ConnectionType().String()

        srv, err := s.SingleServingServer("c2s")
        if err != nil {
                log.Println("Could not start SingleServingServer", err)
                metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "StartSingleServingServer").Inc()
                return record, err
        }

        err = m.SendMessage(protocol.TestPrepare, []byte(strconv.Itoa(srv.Port())))
        if err != nil {
                log.Println("Could not send TestPrepare", err)
                metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestPrepare").Inc()
                return record, err
        }

        testConn, err := srv.ServeOnce(localContext)
        if err != nil {
                log.Println("Could not successfully ServeOnce", err)
                metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "ServeOnce").Inc()
                return record, err
        }

        // When ManageTest exits, close the test connection.
        defer func() {
                // Allow the connection-draining goroutine to empty all buffers in support of
                // poorly-written clients before we close the connection, but do not block the
                // exit of ManageTest on waiting for the test connection to close.
                go func() {
                        time.Sleep(3 * time.Second)
                        warnonerror.Close(testConn, "Could not close test connection")
                }()
        }()

        record.UUID = testConn.UUID()
        record.ServerIP, record.ServerPort = testConn.ServerIPAndPort()
        record.ClientIP, record.ClientPort = testConn.ClientIPAndPort()

        err = m.SendMessage(protocol.TestStart, []byte{})
        if err != nil {
                log.Println("Could not send TestStart", err, record.UUID)
                metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestStart").Inc()
                return record, err
        }

        record.StartTime = time.Now()
        web100Metrics, err := drainForeverButMeasureFor(ctx, testConn, 10*time.Second)
        record.EndTime = time.Now()
        seconds := record.EndTime.Sub(record.StartTime).Seconds()
        log.Println("Ended C2S test on", testConn, record.UUID)
        if err != nil {
                if web100Metrics.TCPInfo.BytesReceived == 0 {
                        log.Println("Could not drain the test connection", err, record.UUID)
                        metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "Drain").Inc()
                        return record, err
                }
                // It is possible for the client to reach 10 seconds slightly before the server does.
                if seconds < 9 {
                        log.Printf("C2S test client only uploaded for %f seconds  %s\n", seconds, record.UUID)
                        metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "EarlyExit").Inc()
                        return record, err
                }
                // More than 9 seconds is fine.
                log.Printf("C2S test had an error (%v) after %f seconds. We will continue with the test.\n", err, seconds)
        }

        throughputValue := 8 * float64(web100Metrics.TCPInfo.BytesReceived) / 1000 / seconds
        record.MeanThroughputMbps = throughputValue / 1000 // Convert Kbps to Mbps

        log.Println(controlConn, "sent us", throughputValue, "Kbps")
        err = m.SendMessage(protocol.TestMsg, []byte(strconv.FormatInt(int64(throughputValue), 10)))
        if err != nil {
                log.Println("Could not send TestMsg with C2S results", err, record.UUID)
                metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestMsg").Inc()
                return record, err
        }

        err = m.SendMessage(protocol.TestFinalize, []byte{})
        if err != nil {
                log.Println("Could not send TestFinalize", err, record.UUID)
                metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestFinalize").Inc()
                return record, err
        }

        return record, nil
}

// drainForeverButMeasureFor is a generic method for draining a connection while
// measuring the connection for the first part of the drain. This method does
// not close the passed-in Connection, and starts a goroutine which runs until
// that Connection is closed.
func drainForeverButMeasureFor(ctx context.Context, conn protocol.MeasuredConnection, d time.Duration) (*web100.Metrics, error) {
        derivedCtx, derivedCancel := context.WithTimeout(ctx, d)
        defer derivedCancel()

        conn.StartMeasuring(derivedCtx)

        errs := make(chan error, 1)
        // This is the "drain forever" part of this function. Read the passed-in
        // connection until the passed-in connection is closed.
        go func() {
                var connErr error
                // Read the connections until the connection is closed. Reading on a closed
                // connection returns an error, which terminates the loop and the goroutine.
                for connErr == nil {
                        _, connErr = conn.ReadBytes()
                }
                errs <- connErr
        }()

        var socketStats *web100.Metrics
        var err error
        select {
        case <-derivedCtx.Done(): // Wait for timeout
                log.Println("Timed out")
                socketStats, err = conn.StopMeasuring()
        case err = <-errs: // Error in c2s transfer
                log.Println("C2S error:", err)
                socketStats, _ = conn.StopMeasuring()
        }
        if socketStats == nil {
                return nil, err
        }
        // socketStats is guaranteed to be non-nil and the TCPInfo element is a value not a pointer.
        return socketStats, err
}

1	package c2s
2
3	import (
4	"context"
5	"log"
6	"strconv"
7	"time"
8
9	"github.com/m-lab/go/warnonerror"
10	"github.com/m-lab/ndt-server/ndt5/metrics"
11	"github.com/m-lab/ndt-server/ndt5/ndt"
12	"github.com/m-lab/ndt-server/ndt5/protocol"
13	"github.com/m-lab/ndt-server/ndt5/web100"
14	)
15
16	// ArchivalData is the data saved by the C2S test. If a researcher wants deeper
17	// data, then they should use the UUID to get deeper data from tcp-info.
18	type ArchivalData struct {
19	// The server and client IP are here as well as in the containing struct
20	// because happy eyeballs means that we may have a IPv4 control connection
21	// causing a IPv6 connection to the test port or vice versa.
22	ServerIP string
23	ServerPort int
24	ClientIP string
25	ClientPort int
26
27	// This is the only field that is really required.
28	UUID string
29
30	// These fields are here to enable analyses that don't require joining with tcp-info data.
31	StartTime time.Time
32	EndTime time.Time
33	MeanThroughputMbps float64
34	// TODO: Add TCPEngine (bbr, cubic, reno, etc.)
35
36	Error string `json:",omitempty"`
37	}
38
39	// ManageTest manages the c2s test lifecycle.
40	func ManageTest(ctx context.Context, controlConn protocol.Connection, s ndt.Server) (record *ArchivalData, err error) {	1✔
41	localContext, localCancel := context.WithTimeout(ctx, 30*time.Second)	1✔
42	defer localCancel()	1✔
43	defer func() {	2✔
44	if err != nil && record != nil {	2✔
45	record.Error = err.Error()	1✔
46	}	1✔
47	}()
48	record = &ArchivalData{}	1✔
49		1✔
50	m := controlConn.Messager()	1✔
51	connType := s.ConnectionType().String()	1✔
52		1✔
53	srv, err := s.SingleServingServer("c2s")	1✔
54	if err != nil {	1✔
55	log.Println("Could not start SingleServingServer", err)	×
56	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "StartSingleServingServer").Inc()	×
57	return record, err	×
58	}	×
59
60	err = m.SendMessage(protocol.TestPrepare, []byte(strconv.Itoa(srv.Port())))	1✔
61	if err != nil {	1✔
62	log.Println("Could not send TestPrepare", err)	×
63	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestPrepare").Inc()	×
64	return record, err	×
65	}	×
66
67	testConn, err := srv.ServeOnce(localContext)	1✔
68	if err != nil {	1✔
69	log.Println("Could not successfully ServeOnce", err)	×
70	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "ServeOnce").Inc()	×
71	return record, err	×
72	}	×
73
74	// When ManageTest exits, close the test connection.
75	defer func() {	2✔
76	// Allow the connection-draining goroutine to empty all buffers in support of	1✔
77	// poorly-written clients before we close the connection, but do not block the	1✔
78	// exit of ManageTest on waiting for the test connection to close.	1✔
79	go func() {	2✔
80	time.Sleep(3 * time.Second)	1✔
81	warnonerror.Close(testConn, "Could not close test connection")	1✔
82	}()	1✔
83	}()
84
85	record.UUID = testConn.UUID()	1✔
86	record.ServerIP, record.ServerPort = testConn.ServerIPAndPort()	1✔
87	record.ClientIP, record.ClientPort = testConn.ClientIPAndPort()	1✔
88		1✔
89	err = m.SendMessage(protocol.TestStart, []byte{})	1✔
90	if err != nil {	1✔
91	log.Println("Could not send TestStart", err, record.UUID)	×
92	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestStart").Inc()	×
93	return record, err	×
94	}	×
95
96	record.StartTime = time.Now()	1✔
97	web100Metrics, err := drainForeverButMeasureFor(ctx, testConn, 10*time.Second)	1✔
98	record.EndTime = time.Now()	1✔
99	seconds := record.EndTime.Sub(record.StartTime).Seconds()	1✔
100	log.Println("Ended C2S test on", testConn, record.UUID)	1✔
101	if err != nil {	2✔
102	if web100Metrics.TCPInfo.BytesReceived == 0 {	1✔
103	log.Println("Could not drain the test connection", err, record.UUID)	×
104	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "Drain").Inc()	×
105	return record, err	×
106	}	×
107	// It is possible for the client to reach 10 seconds slightly before the server does.
108	if seconds < 9 {	2✔
109	log.Printf("C2S test client only uploaded for %f seconds %s\n", seconds, record.UUID)	1✔
110	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "EarlyExit").Inc()	1✔
111	return record, err	1✔
112	}	1✔
113	// More than 9 seconds is fine.
114	log.Printf("C2S test had an error (%v) after %f seconds. We will continue with the test.\n", err, seconds)	1✔
115	}
116
117	throughputValue := 8 * float64(web100Metrics.TCPInfo.BytesReceived) / 1000 / seconds	1✔
118	record.MeanThroughputMbps = throughputValue / 1000 // Convert Kbps to Mbps	1✔
119		1✔
120	log.Println(controlConn, "sent us", throughputValue, "Kbps")	1✔
121	err = m.SendMessage(protocol.TestMsg, []byte(strconv.FormatInt(int64(throughputValue), 10)))	1✔
122	if err != nil {	1✔
123	log.Println("Could not send TestMsg with C2S results", err, record.UUID)	×
124	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestMsg").Inc()	×
125	return record, err	×
126	}	×
127
128	err = m.SendMessage(protocol.TestFinalize, []byte{})	1✔
129	if err != nil {	1✔
130	log.Println("Could not send TestFinalize", err, record.UUID)	×
131	metrics.ClientTestErrors.WithLabelValues(connType, "c2s", "TestFinalize").Inc()	×
132	return record, err	×
133	}	×
134
135	return record, nil	1✔
136	}
137
138	// drainForeverButMeasureFor is a generic method for draining a connection while
139	// measuring the connection for the first part of the drain. This method does
140	// not close the passed-in Connection, and starts a goroutine which runs until
141	// that Connection is closed.
142	func drainForeverButMeasureFor(ctx context.Context, conn protocol.MeasuredConnection, d time.Duration) (*web100.Metrics, error) {	1✔
143	derivedCtx, derivedCancel := context.WithTimeout(ctx, d)	1✔
144	defer derivedCancel()	1✔
145		1✔
146	conn.StartMeasuring(derivedCtx)	1✔
147		1✔
148	errs := make(chan error, 1)	1✔
149	// This is the "drain forever" part of this function. Read the passed-in	1✔
150	// connection until the passed-in connection is closed.	1✔
151	go func() {	2✔
152	var connErr error	1✔
153	// Read the connections until the connection is closed. Reading on a closed	1✔
154	// connection returns an error, which terminates the loop and the goroutine.	1✔
155	for connErr == nil {	2✔
156	_, connErr = conn.ReadBytes()	1✔
157	}	1✔
158	errs <- connErr	1✔
159	}()
160
161	var socketStats *web100.Metrics	1✔
162	var err error	1✔
163	select {	1✔
164	case <-derivedCtx.Done(): // Wait for timeout	1✔
165	log.Println("Timed out")	1✔
166	socketStats, err = conn.StopMeasuring()	1✔
167	case err = <-errs: // Error in c2s transfer	1✔
168	log.Println("C2S error:", err)	1✔
169	socketStats, _ = conn.StopMeasuring()	1✔
170	}
171	if socketStats == nil {	1✔
172	return nil, err	×
173	}	×
174	// socketStats is guaranteed to be non-nil and the TCPInfo element is a value not a pointer.
175	return socketStats, err	1✔
176	}

m-lab / ndt-server / 1298

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous