-
Notifications
You must be signed in to change notification settings - Fork 28
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feat(loggr-syslog-agent): Add https-batch protocol for more efficient…
- Loading branch information
1 parent
0ef8d81
commit 75f9a92
Showing
7 changed files
with
316 additions
and
102 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
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,98 @@ | ||
package syslog | ||
|
||
import ( | ||
"bytes" | ||
"crypto/tls" | ||
"log" | ||
"time" | ||
|
||
"code.cloudfoundry.org/go-loggregator/v10/rpc/loggregator_v2" | ||
metrics "code.cloudfoundry.org/go-metric-registry" | ||
"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress" | ||
) | ||
|
||
const BATCHSIZE = 256 * 1024 | ||
|
||
type HTTPSBatchWriter struct { | ||
HTTPSWriter | ||
msgs chan []byte | ||
batchSize int | ||
sendInterval time.Duration | ||
egrMsgCount float64 | ||
} | ||
|
||
func NewHTTPSBatchWriter( | ||
binding *URLBinding, | ||
netConf NetworkTimeoutConfig, | ||
tlsConf *tls.Config, | ||
egressMetric metrics.Counter, | ||
c *Converter, | ||
) egress.WriteCloser { | ||
client := httpClient(netConf, tlsConf) | ||
binding.URL.Scheme = "https" // reset the scheme for usage to a valid http scheme | ||
BatchWriter := &HTTPSBatchWriter{ | ||
HTTPSWriter: HTTPSWriter{ | ||
url: binding.URL, | ||
appID: binding.AppID, | ||
hostname: binding.Hostname, | ||
client: client, | ||
egressMetric: egressMetric, | ||
syslogConverter: c, | ||
}, | ||
batchSize: BATCHSIZE, | ||
sendInterval: 1 * time.Second, | ||
egrMsgCount: 0, | ||
msgs: make(chan []byte), | ||
} | ||
go BatchWriter.startSender() | ||
return BatchWriter | ||
} | ||
|
||
// Modified Write function | ||
func (w *HTTPSBatchWriter) Write(env *loggregator_v2.Envelope) error { | ||
msgs, err := w.syslogConverter.ToRFC5424(env, w.hostname) | ||
if err != nil { | ||
log.Printf("Failed to parse syslog, dropping faulty message, err: %s", err) | ||
return nil | ||
} | ||
|
||
for _, msg := range msgs { | ||
//There is no correct way of implementing error based retries in the current architecture. | ||
//Retries for https-batching will be implemented at a later point in time. | ||
w.msgs <- msg | ||
} | ||
return nil | ||
} | ||
|
||
func (w *HTTPSBatchWriter) startSender() { | ||
t := time.NewTimer(w.sendInterval) | ||
|
||
var msgBatch bytes.Buffer | ||
var msgCount float64 | ||
reset := func() { | ||
msgBatch.Reset() | ||
msgCount = 0 | ||
t.Reset(w.sendInterval) | ||
} | ||
for { | ||
select { | ||
case msg := <-w.msgs: | ||
length, buffer_err := msgBatch.Write(msg) | ||
if buffer_err != nil { | ||
log.Printf("Failed to write to buffer, dropping buffer of size %d , err: %s", length, buffer_err) | ||
reset() | ||
} else { | ||
msgCount++ | ||
if length >= w.batchSize { | ||
w.sendHttpRequest(msgBatch.Bytes(), msgCount) //nolint:errcheck | ||
reset() | ||
} | ||
} | ||
case <-t.C: | ||
if msgBatch.Len() > 0 { | ||
w.sendHttpRequest(msgBatch.Bytes(), msgCount) //nolint:errcheck | ||
reset() | ||
} | ||
} | ||
} | ||
} |
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,135 @@ | ||
package syslog_test | ||
|
||
import ( | ||
"bytes" | ||
"crypto/tls" | ||
"io" | ||
"net/http" | ||
"net/http/httptest" | ||
"time" | ||
|
||
"code.cloudfoundry.org/go-loggregator/v10/rfc5424" | ||
"code.cloudfoundry.org/go-loggregator/v10/rpc/loggregator_v2" | ||
metricsHelpers "code.cloudfoundry.org/go-metric-registry/testhelpers" | ||
"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress" | ||
"code.cloudfoundry.org/loggregator-agent-release/src/pkg/egress/syslog" | ||
. "github.com/onsi/ginkgo/v2" | ||
. "github.com/onsi/gomega" | ||
) | ||
|
||
var string_to_1024_chars = "saljdflajsdssdfsdfljkfkajafjajlköflkjöjaklgljksdjlakljkflkjweljklkwjejlkfekljwlkjefjklwjklsdajkljklwerlkaskldgjksakjekjwrjkljasdjkgfkljwejklrkjlklasdkjlsadjlfjlkadfljkajklsdfjklslkdfjkllkjasdjkflsdlakfjklasldfkjlasdjfkjlsadlfjklaljsafjlslkjawjklerkjljklasjkdfjklwerjljalsdjkflwerjlkwejlkarjklalkklfsdjlfhkjsdfkhsewhkjjasdjfkhwkejrkjahjefkhkasdjhfkashfkjwehfkksadfjaskfkhjdshjfhewkjhasdfjdajskfjwehkfajkankaskjdfasdjhfkkjhjjkasdfjhkjahksdf" | ||
|
||
var _ = Describe("HTTPS_batch", func() { | ||
var ( | ||
netConf syslog.NetworkTimeoutConfig | ||
skipSSLTLSConfig = &tls.Config{ | ||
InsecureSkipVerify: true, //nolint:gosec | ||
} | ||
c = syslog.NewConverter() | ||
drain *SpyDrain | ||
b *syslog.URLBinding | ||
writer egress.WriteCloser | ||
) | ||
string_to_1024_chars += string_to_1024_chars | ||
|
||
BeforeEach(func() { | ||
drain = newBatchMockDrain(200) | ||
b = buildURLBinding( | ||
drain.URL, | ||
"test-app-id", | ||
"test-hostname", | ||
) | ||
writer = syslog.NewHTTPSBatchWriter( | ||
b, | ||
netConf, | ||
skipSSLTLSConfig, | ||
&metricsHelpers.SpyMetric{}, | ||
c, | ||
) | ||
}) | ||
|
||
It("testing simple appending of one log", func() { | ||
env1 := buildLogEnvelope("APP", "1", "message 1", loggregator_v2.Log_OUT) | ||
Expect(writer.Write(env1)).To(Succeed()) | ||
env2 := buildLogEnvelope("APP", "2", "message 2", loggregator_v2.Log_OUT) | ||
Expect(writer.Write(env2)).To(Succeed()) | ||
time.Sleep(1050 * time.Millisecond) | ||
|
||
Expect(drain.getMessagesSize()).Should(Equal(2)) | ||
expected := &rfc5424.Message{ | ||
AppName: "test-app-id", | ||
Hostname: "test-hostname", | ||
Priority: rfc5424.Priority(14), | ||
ProcessID: "[APP/1]", | ||
Message: []byte("message 1\n"), | ||
} | ||
Expect(drain.messages[0].AppName).To(Equal(expected.AppName)) | ||
Expect(drain.messages[0].Hostname).To(Equal(expected.Hostname)) | ||
Expect(drain.messages[0].Priority).To(BeEquivalentTo(expected.Priority)) | ||
Expect(drain.messages[0].ProcessID).To(Equal(expected.ProcessID)) | ||
Expect(drain.messages[0].Message).To(Equal(expected.Message)) | ||
expected = &rfc5424.Message{ | ||
AppName: "test-app-id", | ||
Hostname: "test-hostname", | ||
Priority: rfc5424.Priority(14), | ||
ProcessID: "[APP/2]", | ||
Message: []byte("message 2\n"), | ||
} | ||
Expect(drain.messages[1].AppName).To(Equal(expected.AppName)) | ||
Expect(drain.messages[1].Hostname).To(Equal(expected.Hostname)) | ||
Expect(drain.messages[1].Priority).To(BeEquivalentTo(expected.Priority)) | ||
Expect(drain.messages[1].ProcessID).To(Equal(expected.ProcessID)) | ||
Expect(drain.messages[1].Message).To(Equal(expected.Message)) | ||
}) | ||
|
||
It("test batch dispatching with all logs in a given timeframe", func() { | ||
env1 := buildLogEnvelope("APP", "1", "string to get log to 1024 characters:"+string_to_1024_chars, loggregator_v2.Log_OUT) | ||
for i := 0; i < 10; i++ { | ||
Expect(writer.Write(env1)).To(Succeed()) | ||
time.Sleep(99 * time.Millisecond) | ||
} | ||
Expect(drain.getMessagesSize()).Should(Equal(0)) | ||
time.Sleep(100 * time.Millisecond) | ||
Expect(drain.getMessagesSize()).Should(Equal(10)) | ||
}) | ||
|
||
It("probabilistic test for race condition", func() { | ||
env1 := buildLogEnvelope("APP", "1", "string to get log to 1024 characters:"+string_to_1024_chars, loggregator_v2.Log_OUT) | ||
for i := 0; i < 10; i++ { | ||
Expect(writer.Write(env1)).To(Succeed()) | ||
time.Sleep(99 * time.Millisecond) | ||
} | ||
time.Sleep(100 * time.Millisecond) | ||
Expect(drain.getMessagesSize()).Should(Equal(10)) | ||
}) | ||
}) | ||
|
||
func newBatchMockDrain(status int) *SpyDrain { | ||
drain := &SpyDrain{} | ||
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | ||
|
||
body, err := io.ReadAll(r.Body) | ||
Expect(err).ToNot(HaveOccurred()) | ||
defer r.Body.Close() | ||
|
||
println(body) | ||
|
||
message := &rfc5424.Message{} | ||
|
||
messages := bytes.SplitAfter(body, []byte("\n")) | ||
for _, raw := range messages { | ||
if bytes.Equal(raw, []byte("")) { | ||
continue | ||
} | ||
message = &rfc5424.Message{} | ||
err = message.UnmarshalBinary(raw) | ||
Expect(err).ToNot(HaveOccurred()) | ||
drain.appendMessage(message) | ||
drain.appendHeader(r.Header) | ||
} | ||
w.WriteHeader(status) | ||
}) | ||
server := httptest.NewTLSServer(handler) | ||
drain.Server = server | ||
return drain | ||
} |
Oops, something went wrong.