Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Using specific pass for each host in redis output #16206

Merged
merged 21 commits into from
Feb 24, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
0d9cb32
Using specific pass for each host in redis output
rvillablanca Feb 9, 2020
b88f2a7
Merge remote-tracking branch 'origin/master' into specific-password-r…
rvillablanca Feb 13, 2020
02fb2b1
Partials advances parsing redis scheme
rvillablanca Feb 13, 2020
abb0df4
Validate tls config when schema is 'rediss'
rvillablanca Feb 13, 2020
3f23756
Fix collition name with package url
rvillablanca Feb 13, 2020
9ce4c41
* fix bug with host
rvillablanca Feb 13, 2020
79ad5a2
More validations in url
rvillablanca Feb 14, 2020
a37d524
Merge remote-tracking branch 'origin/master' into specific-password-r…
rvillablanca Feb 14, 2020
b219e24
Merge remote-tracking branch 'origin/master' into specific-password-r…
rvillablanca Feb 16, 2020
56cfc0e
Add some corrections
rvillablanca Feb 16, 2020
f8d1161
Merge remote-tracking branch 'origin/master' into specific-password-r…
rvillablanca Feb 20, 2020
2d9e68b
Typo
rvillablanca Feb 20, 2020
7e53c61
unit test and integration tests
rvillablanca Feb 21, 2020
877f1fa
Merge remote-tracking branch 'origin/master' into specific-password-r…
rvillablanca Feb 21, 2020
ed763da
Fix redis output integration tests
rvillablanca Feb 21, 2020
2296621
Minor improvements when testing redis output
rvillablanca Feb 21, 2020
30a2bf0
Add license header
rvillablanca Feb 22, 2020
df66485
Add changes to changelog
rvillablanca Feb 22, 2020
87106f9
Fix license header
rvillablanca Feb 22, 2020
0bd9c96
Fix imports
rvillablanca Feb 22, 2020
831348c
Formatting using make fmt update
rvillablanca Feb 22, 2020
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.next.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
- Add document_id setting to decode_json_fields processor. {pull}15859[15859]
- Include network information by default on add_host_metadata and add_observer_metadata. {issue}15347[15347] {pull}16077[16077]
- Add `aws_ec2` provider for autodiscover. {issue}12518[12518] {pull}14823[14823]
- Add support for multiple password in redis output. {issue}16058[16058] {pull}16206[16206]

*Auditbeat*

Expand Down
63 changes: 52 additions & 11 deletions libbeat/outputs/redis/redis.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,9 @@ package redis

import (
"errors"
"fmt"
"net/url"
"strings"
"time"

"github.com/elastic/beats/libbeat/beat"
Expand All @@ -42,6 +45,8 @@ const (
defaultWaitRetry = 1 * time.Second
defaultMaxWaitRetry = 60 * time.Second
defaultPort = 6379
redisScheme = "redis"
tlsRedisScheme = "rediss"
)

func init() {
Expand Down Expand Up @@ -110,27 +115,63 @@ func makeRedis(
return outputs.Fail(err)
}

transp := &transport.Config{
Timeout: config.Timeout,
Proxy: &config.Proxy,
TLS: tls,
Stats: observer,
}

clients := make([]outputs.NetworkClient, len(hosts))
for i, host := range hosts {
enc, err := codec.CreateEncoder(beat, config.Codec)
for i, h := range hosts {
hasScheme := true
if parts := strings.SplitN(h, "://", 2); len(parts) != 2 {
h = fmt.Sprintf("%s://%s", redisScheme, h)
hasScheme = false
}

hostUrl, err := url.Parse(h)
if err != nil {
return outputs.Fail(err)
}

if hostUrl.Host == "" {
return outputs.Fail(fmt.Errorf("invalid redis url host %s", hostUrl.Host))
}

if hostUrl.Scheme != redisScheme && hostUrl.Scheme != tlsRedisScheme {
return outputs.Fail(fmt.Errorf("invalid redis url scheme %s", hostUrl.Scheme))
}

transp := &transport.Config{
Timeout: config.Timeout,
Proxy: &config.Proxy,
TLS: tls,
Stats: observer,
}
rvillablanca marked this conversation as resolved.
Show resolved Hide resolved

switch hostUrl.Scheme {
case redisScheme:
if hasScheme {
transp.TLS = nil // disable TLS if user explicitely set `redis` scheme
}
case tlsRedisScheme:
if transp.TLS == nil {
transp.TLS = &transport.TLSConfig{} // enable with system default if TLS was not configured
}
}
rvillablanca marked this conversation as resolved.
Show resolved Hide resolved

conn, err := transport.NewClient(transp, "tcp", hostUrl.Host, defaultPort)
if err != nil {
return outputs.Fail(err)
}

conn, err := transport.NewClient(transp, "tcp", host, defaultPort)
pass := config.Password
hostPass, passSet := hostUrl.User.Password()
if passSet {
pass = hostPass
}
urso marked this conversation as resolved.
Show resolved Hide resolved

enc, err := codec.CreateEncoder(beat, config.Codec)
if err != nil {
return outputs.Fail(err)
}

client := newClient(conn, observer, config.Timeout,
config.Password, config.Db, key, dataType, config.Index, enc)
pass, config.Db, key, dataType, config.Index, enc)
clients[i] = newBackoffClient(client, config.Backoff.Init, config.Backoff.Max)
}

Expand Down
42 changes: 40 additions & 2 deletions libbeat/outputs/redis/redis_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,7 @@ const (
)

func TestPublishListTCP(t *testing.T) {
key := "test_publist_tcp"
key := "test_publish_tcp"
db := 0
redisConfig := map[string]interface{}{
"hosts": []string{getRedisAddr()},
Expand All @@ -67,7 +67,7 @@ func TestPublishListTCP(t *testing.T) {
}

func TestPublishListTLS(t *testing.T) {
key := "test_publist_tls"
key := "test_publish_tls"
db := 0
redisConfig := map[string]interface{}{
"hosts": []string{getSRedisAddr()},
Expand All @@ -85,6 +85,44 @@ func TestPublishListTLS(t *testing.T) {
testPublishList(t, redisConfig)
}

func TestWithSchema(t *testing.T) {
redisURL := "redis://" + getRedisAddr()
sredisURL := "rediss://" + getSRedisAddr()

cases := map[string]struct {
host string
}{
"redis ignores ssl settings": {
host: redisURL,
},
"sredis schema sends via tls": {
host: sredisURL,
},
}

for name, test := range cases {
t.Run(name, func(t *testing.T) {
key := "test_publish_tls"
db := 0
redisConfig := map[string]interface{}{
"hosts": []string{test.host},
"key": key,
"db": db,
"datatype": "list",
"timeout": "5s",

"ssl.verification_mode": "full",
"ssl.certificate_authorities": []string{
"../../../testing/environments/docker/sredis/pki/tls/certs/sredis.crt",
},
}

testPublishList(t, redisConfig)
})
}

}
rvillablanca marked this conversation as resolved.
Show resolved Hide resolved

func testPublishList(t *testing.T, cfg map[string]interface{}) {
batches := 100
batchSize := 1000
Expand Down
120 changes: 120 additions & 0 deletions libbeat/outputs/redis/redis_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. 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 redis

import (
"testing"

"github.com/stretchr/testify/assert"

"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/outputs"
_ "github.com/elastic/beats/libbeat/outputs/codec/json"
)

type checker func(*testing.T, outputs.Group)

func checks(cs ...checker) checker {
return func(t *testing.T, g outputs.Group) {
for _, c := range cs {
c(t, g)
}
}
}

func clientsLen(required int) checker {
return func(t *testing.T, group outputs.Group) {
assert.Len(t, group.Clients, required)
}
}

func clientPassword(index int, pass string) checker {
return func(t *testing.T, group outputs.Group) {
redisClient := group.Clients[index].(*backoffClient)
assert.Equal(t, redisClient.client.password, pass)
}
}

func TestMakeRedis(t *testing.T) {
tests := map[string]struct {
config map[string]interface{}
valid bool
checks checker
}{
rvillablanca marked this conversation as resolved.
Show resolved Hide resolved
"no host": {
config: map[string]interface{}{
"hosts": []string{},
},
},
"invald scheme": {
config: map[string]interface{}{
"hosts": []string{"redisss://localhost:6379"},
},
},
"Single host": {
config: map[string]interface{}{
"hosts": []string{"localhost:6379"},
},
valid: true,
checks: checks(clientsLen(1), clientPassword(0, "")),
},
"Multiple hosts": {
config: map[string]interface{}{
"hosts": []string{"redis://localhost:6379", "rediss://localhost:6380"},
},
valid: true,
checks: clientsLen(2),
},
"Default password": {
config: map[string]interface{}{
"hosts": []string{"redis://localhost:6379"},
"password": "defaultPassword",
},
valid: true,
checks: checks(clientsLen(1), clientPassword(0, "defaultPassword")),
},
"Specific and default password": {
config: map[string]interface{}{
"hosts": []string{"redis://localhost:6379", "rediss://:mypassword@localhost:6380"},
"password": "defaultPassword",
},
valid: true,
checks: checks(
clientsLen(2),
clientPassword(0, "defaultPassword"),
clientPassword(1, "mypassword"),
),
},
}
beatInfo := beat.Info{Beat: "libbeat", Version: "1.2.3"}
for name, test := range tests {
t.Run(name, func(t *testing.T) {
cfg, err := common.NewConfigFrom(test.config)
assert.NoError(t, err)
groups, err := makeRedis(nil, beatInfo, outputs.NewNilObserver(), cfg)
assert.Equal(t, err == nil, test.valid)
if err != nil && test.valid {
t.Log(err)
}
if test.checks != nil {
test.checks(t, groups)
}
})
}
}