Cassandra Connect Support (#5952)
* Init cassandra * Init cassandra * Initial Cassandra Command Support For cqlsh * Add Make Gen * adding cassandra test * Cleanup cassandra docker setup * Adding container wait * Container restart * Minor Fix * Cleanup docker and test * Adding final tests * Removing comment * Fixing cleanup (uncomment) * Removing bad yaml * Cleanup * Cleanup * Adressing PR review comments * Adding test support * fix spacing * Adding changelog * docs * docs * adding cassandra page * nav * Versbose commenting * cleanup naming of file * updating helper func to return errors * doc change * Removed an incorrect comment and now passing the config by reference. * Comment cleanup * Fix option warning in cassandra connect command (#5951) * Add healthcheck to caller function as it's resposible for container health * Resolve Comments * Update Log Typo --------- Co-authored-by: Bharath Gajjala <120367134+bgajjala8@users.noreply.github.com>
This commit is contained in:
Ryan DerrandBharath Gajjala authored and 
GitHub committed
1 parent
029aa361bd
commit
4d3cb76c28
10 files changed
+504
-1
No files matched your search
@@ -8,6 +8,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
@@ -26,6 +27,14 @@ type Container struct {
|
||||
UriNetwork string
|
||||
}
|
||||
|
||||
// cassandraConfig stores configuration details for the Cassandra container
|
||||
type cassandraConfig struct {
|
||||
User string
|
||||
Password string
|
||||
Keyspace string
|
||||
NetworkAlias string
|
||||
}
|
||||
|
||||
// StartBoundaryDatabase spins up a postgres database in a docker container.
|
||||
// Returns information about the container
|
||||
func StartBoundaryDatabase(t testing.TB, pool *dockertest.Pool, network *dockertest.Network, repository, tag string) *Container {
|
||||
@@ -344,3 +353,129 @@ func StartMysql(t testing.TB, pool *dockertest.Pool, network *dockertest.Network
|
||||
UriNetwork: fmt.Sprintf("mysql://%s:%s@%s:3306/%s", mysqlUser, mysqlPassword, networkAlias, mysqlDb),
|
||||
}
|
||||
}
|
||||
|
||||
// StartCassandra starts a Cassandra database in a docker container.
|
||||
// Returns information about the container
|
||||
func StartCassandra(t testing.TB, pool *dockertest.Pool, network *dockertest.Network, repository, tag string) *Container {
|
||||
t.Log("Starting Cassandra database...")
|
||||
c, err := LoadConfig()
|
||||
require.NoError(t, err)
|
||||
|
||||
err = pool.Client.PullImage(docker.PullImageOptions{
|
||||
Repository: fmt.Sprintf("%s/%s", c.DockerMirror, repository),
|
||||
Tag: tag,
|
||||
}, docker.AuthConfiguration{})
|
||||
require.NoError(t, err)
|
||||
|
||||
config := cassandraConfig{
|
||||
User: "e2eboundary",
|
||||
Password: "e2eboundary",
|
||||
Keyspace: "e2eboundarykeyspace",
|
||||
NetworkAlias: "e2ecassandra",
|
||||
}
|
||||
|
||||
resource, err := pool.RunWithOptions(&dockertest.RunOptions{
|
||||
Repository: fmt.Sprintf("%s/%s", c.DockerMirror, repository),
|
||||
Tag: tag,
|
||||
Env: []string{
|
||||
"CASSANDRA_CLUSTER_NAME=e2e-boundary-cluster",
|
||||
},
|
||||
|
||||
ExposedPorts: []string{"9042/tcp"},
|
||||
Name: config.NetworkAlias,
|
||||
Networks: []*dockertest.Network{network},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Cassandra container takes a while to start due to the gossip protocol needing to settle and establish connections.
|
||||
// This relies on the pool's extended maxWait time to ensure the container is healthy.
|
||||
err = pool.Retry(func() error {
|
||||
cmd := exec.Command("docker", "exec", config.NetworkAlias, "cqlsh", "-e", "SELECT now() FROM system.local;")
|
||||
output, cmdErr := cmd.CombinedOutput()
|
||||
if cmdErr != nil {
|
||||
return fmt.Errorf("failed to connect to Cassandra container '%s': %v\nOutput: %s", config.NetworkAlias, cmdErr, string(output))
|
||||
}
|
||||
return nil
|
||||
})
|
||||
require.NoError(t, err, "Cassandra container did not start in time or is not healthy")
|
||||
|
||||
err = setupCassandraAuthAndUser(t, resource, pool, &config)
|
||||
require.NoError(t, err)
|
||||
|
||||
return &Container{
|
||||
Resource: resource,
|
||||
UriLocalhost: fmt.Sprintf(
|
||||
"cassandra://%s:%s@%s/%s",
|
||||
config.User,
|
||||
config.Password,
|
||||
resource.GetHostPort("9042/tcp"),
|
||||
config.Keyspace,
|
||||
),
|
||||
UriNetwork: fmt.Sprintf(
|
||||
"cassandra://%s:%s@%s:9042/%s",
|
||||
config.User,
|
||||
config.Password,
|
||||
config.NetworkAlias,
|
||||
config.Keyspace,
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
// setupCassandraAuthAndUser enables authentication on a Cassandra container and creates a user with permissions.
|
||||
func setupCassandraAuthAndUser(t testing.TB, resource *dockertest.Resource, pool *dockertest.Pool, config *cassandraConfig) error {
|
||||
t.Helper()
|
||||
t.Log("Configuring Cassandra authentication and user permissions...")
|
||||
|
||||
t.Logf("Initializing Cassandra keyspace: %s...", config.Keyspace)
|
||||
createKeyspaceCmd := fmt.Sprintf(
|
||||
"CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};",
|
||||
config.Keyspace,
|
||||
)
|
||||
if err := exec.Command("docker", "exec", config.NetworkAlias, "cqlsh", "-e", createKeyspaceCmd).Run(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Commands to enable authentication and authorization by editing cassandra.yaml
|
||||
sedCmd := []string{
|
||||
"sed", "-i",
|
||||
"-e", "s/^authenticator:.*/authenticator: PasswordAuthenticator/",
|
||||
"-e", "s/^authorizer:.*/authorizer: CassandraAuthorizer/",
|
||||
"-e", "s/^role_manager:.*/role_manager: CassandraRoleManager/",
|
||||
"/etc/cassandra/cassandra.yaml",
|
||||
}
|
||||
if _, err := resource.Exec(sedCmd, dockertest.ExecOptions{}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := pool.Client.RestartContainer(resource.Container.ID, uint(pool.MaxWait.Seconds())); err != nil {
|
||||
return err
|
||||
}
|
||||
t.Log("Waiting for Cassandra container to restart and apply authentication settings...")
|
||||
|
||||
// Wait for Cassandra to be up with authentication enabled
|
||||
if err := pool.Retry(func() error {
|
||||
return exec.Command(
|
||||
"docker", "exec", config.NetworkAlias,
|
||||
"cqlsh", "-u", "cassandra", "-p", "cassandra",
|
||||
"-e", "SELECT now() FROM system.local;",
|
||||
).Run()
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
t.Log("Creating Cassandra user and granting permissions...")
|
||||
cqlCmds := []string{
|
||||
fmt.Sprintf("CREATE ROLE IF NOT EXISTS %s WITH PASSWORD = '%s' AND LOGIN = true;", config.User, config.Password),
|
||||
fmt.Sprintf("GRANT ALL PERMISSIONS ON KEYSPACE %s TO %s;", config.Keyspace, config.User),
|
||||
fmt.Sprintf("USE %s; CREATE TABLE IF NOT EXISTS users (id UUID PRIMARY KEY, name TEXT, created_at TIMESTAMP);", config.Keyspace),
|
||||
}
|
||||
for _, cmd := range cqlCmds {
|
||||
if err := exec.Command(
|
||||
"docker", "exec", config.NetworkAlias,
|
||||
"cqlsh", "-u", "cassandra", "-p", "cassandra", "-e", cmd,
|
||||
).Run(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: BUSL-1.1
|
||||
|
||||
package base_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/url"
|
||||
"os/exec"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/creack/pty"
|
||||
"github.com/hashicorp/boundary/internal/target"
|
||||
"github.com/hashicorp/boundary/testing/internal/e2e"
|
||||
"github.com/hashicorp/boundary/testing/internal/e2e/boundary"
|
||||
"github.com/hashicorp/boundary/testing/internal/e2e/infra"
|
||||
"github.com/ory/dockertest/v3"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestCliTcpTargetConnectCassandra uses the boundary cli to connect to a
|
||||
// target using `connect cassandra`
|
||||
func TestCliTcpTargetConnectCassandra(t *testing.T) {
|
||||
e2e.MaybeSkipTest(t)
|
||||
|
||||
pool, err := dockertest.NewPool("")
|
||||
require.NoError(t, err)
|
||||
|
||||
// Increase timeout to accommodate Cassandra's longer startup duration due to gossip needing to settle
|
||||
pool.MaxWait = 90 * time.Second
|
||||
ctx := context.Background()
|
||||
|
||||
// e2e_cluster network is created by the e2e infra setup
|
||||
network, err := pool.NetworksByName("e2e_cluster")
|
||||
require.NoError(t, err, "Failed to get e2e_cluster network")
|
||||
|
||||
c := infra.StartCassandra(t, pool, &network[0], "cassandra", "latest")
|
||||
require.NotNil(t, c, "Cassandra container should not be nil")
|
||||
t.Cleanup(func() {
|
||||
if err := pool.Purge(c.Resource); err != nil {
|
||||
t.Logf("Failed to purge Cassandra container: %v", err)
|
||||
}
|
||||
})
|
||||
|
||||
u, err := url.Parse(c.UriNetwork)
|
||||
t.Log(u)
|
||||
require.NoError(t, err, "Failed to parse Cassandra URL")
|
||||
|
||||
user, hostname, port, keyspace := u.User.Username(), u.Hostname(), u.Port(), u.Path[1:]
|
||||
pw, pwSet := u.User.Password()
|
||||
|
||||
t.Logf("Cassandra info: user=%s, keyspace=%s, host=%s, port=%s, password-set:%t",
|
||||
user, keyspace, hostname, port, pwSet)
|
||||
|
||||
boundary.AuthenticateAdminCli(t, ctx)
|
||||
|
||||
orgId, err := boundary.CreateOrgCli(t, ctx)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
ctx := context.Background()
|
||||
boundary.AuthenticateAdminCli(t, ctx)
|
||||
output := e2e.RunCommand(ctx, "boundary", e2e.WithArgs("scopes", "delete", "-id", orgId))
|
||||
require.NoError(t, output.Err, string(output.Stderr))
|
||||
})
|
||||
|
||||
projectId, err := boundary.CreateProjectCli(t, ctx, orgId)
|
||||
require.NoError(t, err)
|
||||
|
||||
targetId, err := boundary.CreateTargetCli(
|
||||
t,
|
||||
ctx,
|
||||
projectId,
|
||||
port,
|
||||
target.WithAddress(hostname),
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
storeId, err := boundary.CreateCredentialStoreStaticCli(t, ctx, projectId)
|
||||
require.NoError(t, err)
|
||||
|
||||
credentialId, err := boundary.CreateStaticCredentialPasswordCli(
|
||||
t,
|
||||
ctx,
|
||||
storeId,
|
||||
user,
|
||||
pw,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
err = boundary.AddBrokeredCredentialSourceToTargetCli(t, ctx, targetId, credentialId)
|
||||
require.NoError(t, err)
|
||||
|
||||
cmd := exec.CommandContext(ctx,
|
||||
"boundary",
|
||||
"connect", "cassandra",
|
||||
"-target-id", targetId,
|
||||
"-keyspace", keyspace,
|
||||
)
|
||||
f, err := pty.Start(cmd)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() {
|
||||
err := f.Close()
|
||||
require.NoError(t, err)
|
||||
})
|
||||
|
||||
_, err = f.Write([]byte("DESCRIBE KEYSPACES;\n"))
|
||||
require.NoError(t, err)
|
||||
_, err = f.Write([]byte(fmt.Sprintf("SELECT keyspace_name FROM system_schema.keyspaces WHERE keyspace_name = '%s';\n", keyspace)))
|
||||
require.NoError(t, err)
|
||||
_, err = f.Write([]byte("exit\n"))
|
||||
require.NoError(t, err)
|
||||
|
||||
var buf bytes.Buffer
|
||||
_, _ = io.Copy(&buf, f)
|
||||
|
||||
output := buf.String()
|
||||
t.Logf("Cassandra session output: %s", output)
|
||||
|
||||
require.Contains(t, output, "keyspace_name")
|
||||
require.Contains(t, output, keyspace)
|
||||
|
||||
t.Log("Successfully connected to Cassandra target")
|
||||
}
|
||||
Reference in new issue
Block a user