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

Node: add AZ Affinity ReadFrom strategy Support #2686

Merged
merged 10 commits into from
Nov 20, 2024
15 changes: 15 additions & 0 deletions node/src/BaseClient.ts
Original file line number Diff line number Diff line change
Expand Up @@ -574,6 +574,19 @@ export interface BaseClientConfiguration {
* used.
*/
inflightRequestsLimit?: number;
/**
* Availability Zone of the client.
* This setting ensures that read operations are directed to nodes within the specified AZ.
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • If ReadFrom strategy is AZAffinity, this setting ensures that readonly commands are directed to replicas within the specified AZ if exits.

* If not set, the AZAffinity strategy will not be applied, and read operations will follow the selected ReadFrom strategy without AZ-based routing.
*
* @example
* ```typescript
* // Example configuration for setting client availability zone and read strategy
* configuration.clientAz = 'us-east-1a'; // Sets the client's availability zone
* configuration.readFrom = 'AZAffinity'; // Directs read operations to nodes within the same AZ
* ```
*/
clientAz?: string;
}

/**
Expand Down Expand Up @@ -719,6 +732,7 @@ export class BaseClient {
private readonly pubsubFutures: [PromiseFunction, ErrorFunction][] = [];
private pendingPushNotification: response.Response[] = [];
private readonly inflightRequestsLimit: number;
private readonly clientAz: string | undefined;
private config: BaseClientConfiguration | undefined;

protected configurePubsub(
Expand Down Expand Up @@ -7578,6 +7592,7 @@ export class BaseClient {
readFrom,
authenticationInfo,
inflightRequestsLimit: options.inflightRequestsLimit,
clientAz: options.clientAz ?? null,
};
}

Expand Down
271 changes: 270 additions & 1 deletion node/tests/GlideClient.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import {
GlideString,
ListDirection,
ProtocolVersion,
ReadFrom,
RequestError,
Script,
Transaction,
Expand Down Expand Up @@ -51,7 +52,9 @@ const TIMEOUT = 50000;
describe("GlideClient", () => {
let testsFailed = 0;
let cluster: ValkeyCluster;
let azCluster: ValkeyCluster;
let client: GlideClient;
let azClient: GlideClient;
beforeAll(async () => {
const standaloneAddresses = global.STAND_ALONE_ENDPOINT;
cluster = standaloneAddresses
Expand All @@ -61,17 +64,28 @@ describe("GlideClient", () => {
getServerVersion,
)
: await ValkeyCluster.createCluster(false, 1, 1, getServerVersion);

azCluster = standaloneAddresses
? await ValkeyCluster.initFromExistingCluster(
false,
parseEndpoints(standaloneAddresses),
getServerVersion,
)
: await ValkeyCluster.createCluster(false, 1, 1, getServerVersion);
}, 20000);

afterEach(async () => {
await flushAndCloseClient(false, cluster.getAddresses(), client);
await flushAndCloseClient(false, azCluster.getAddresses(), azClient);
});

afterAll(async () => {
if (testsFailed === 0) {
await cluster.close();
await azCluster.close()
} else {
await cluster.close(true);
await azCluster.close()
}
}, TIMEOUT);

Expand Down Expand Up @@ -1500,18 +1514,273 @@ describe("GlideClient", () => {
}
},
);
describe("GlideClient - AZAffinity Read Strategy Test", () => {
it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
"should route GET commands to all replicas with the same AZ using protocol %p",
async (protocol) => {
const az = "us-east-1a";
const GET_CALLS = 3;
const get_cmdstat = `cmdstat_get:calls=${GET_CALLS}`;

let client_for_config_set;
let client_for_testing_az;

try {
// Stage 1: Configure nodes
client_for_config_set = await GlideClient.createClient(
getClientConfigurationOption(
azCluster.getAddresses(),
protocol,
),
);

// Skip test if version is below 8.0.0
if (azCluster.checkIfServerVersionLessThan("8.0.0")) {
console.log(
"Skipping test: requires Valkey 8.0.0 or higher",
);
return;
}

await client_for_config_set.customCommand([
"CONFIG",
"RESETSTAT",
]);

await client_for_config_set.customCommand([
"CONFIG",
"SET",
"availability-zone",
az,
]);

// Stage 2: Create AZ affinity client and verify configuration
client_for_testing_az = await GlideClient.createClient(
getClientConfigurationOption(
azCluster.getAddresses(),
protocol,
{
readFrom: "AZAffinity" as ReadFrom,
clientAz: az,
},
),
);

const azs = await client_for_testing_az.customCommand([
"CONFIG",
"GET",
"availability-zone",
]);

if (Array.isArray(azs) && azs.length > 0) {
const configItem = azs[0] as {
key: string;
value: string;
};

if (
configItem &&
configItem.key === "availability-zone"
) {
expect(configItem.value).toBe(az);
} else {
throw new Error("Invalid config item format");
}
} else {
throw new Error(
"Unexpected response format from CONFIG GET command",
);
}

// Stage 3: Set test data and perform GET operations
await client_for_testing_az.set("foo", "testvalue");

for (let i = 0; i < GET_CALLS; i++) {
await client_for_testing_az.get("foo");
}

// Stage 4: Verify GET commands were routed correctly
const info_result =
await client_for_testing_az.customCommand([
"INFO",
"COMMANDSTATS",
]);

if (
typeof info_result === "string" &&
info_result.includes(get_cmdstat)
) {
expect(info_result).toContain(get_cmdstat);
} else {
throw new Error(
"Unexpected response format from INFO command in standalone mode",
);
}
} finally {
// Cleanup
await client_for_config_set?.close();
await client_for_testing_az?.close();
}
},
);
});
describe("GlideClient - AZAffinity Routing to 1 replica", () => {
it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
"should route commands to single replica with AZ using protocol %p",
async (protocol) => {
const az = "us-east-1a";
const GET_CALLS = 3;
const get_cmdstat = `calls=${GET_CALLS}`;

let client_for_config_set;
let client_for_testing_az;

try {
// Stage 1: Configure nodes
client_for_config_set = await GlideClient.createClient(
getClientConfigurationOption(
azCluster.getAddresses(),
protocol,
),
);

// Skip test if version is below 8.0.0
if (azCluster.checkIfServerVersionLessThan("8.0.0")) {
console.log(
"Skipping test: requires Valkey 8.0.0 or higher",
);
return;
}

await client_for_config_set.customCommand([
"CONFIG",
"SET",
"availability-zone",
az,
]);

await client_for_config_set.customCommand([
"CONFIG",
"RESETSTAT",
]);

// Stage 2: Create AZ affinity client and verify configuration
client_for_testing_az = await GlideClient.createClient(
getClientConfigurationOption(
azCluster.getAddresses(),
protocol,
{
readFrom: "AZAffinity",
clientAz: az,
},
),
);
await client_for_testing_az.set("foo", "testvalue");

for (let i = 0; i < GET_CALLS; i++) {
await client_for_testing_az.get("foo");
}

// Stage 4: Verify GET commands were routed correctly
const info_result =
await client_for_testing_az.customCommand([
"INFO",
"ALL",
]);

if (typeof info_result === "string") {
expect(info_result).toContain(get_cmdstat);
expect(info_result).toContain(
`availability_zone:${az}`,
);
} else {
throw new Error(
"Unexpected response format from INFO COMMANDSTATS command",
);
}
} finally {
await client_for_config_set?.close();
await client_for_testing_az?.close();
}
},
);
});
describe("GlideClient - AZAffinity with Non-existing AZ", () => {
it.each([ProtocolVersion.RESP2, ProtocolVersion.RESP3])(
"should route commands to a replica when AZ does not exist using protocol %p",
async (protocol) => {
const GET_CALLS = 4;
const get_cmdstat = `cmdstat_get:calls=${GET_CALLS}`;

let client_for_testing_az;

try {
if (azCluster.checkIfServerVersionLessThan("8.0.0")) {
console.log(
"Skipping test: requires Valkey 8.0.0 or higher",
);
return;
}

client_for_testing_az = await GlideClient.createClient(
getClientConfigurationOption(
azCluster.getAddresses(),
protocol,
{
readFrom: "AZAffinity",
clientAz: "non-existing-az",
requestTimeout: 2000,
},
),
);

await client_for_testing_az.customCommand([
"CONFIG",
"RESETSTAT",
]);

for (let i = 0; i < GET_CALLS; i++) {
await client_for_testing_az.get("foo");
}

const info_result =
await client_for_testing_az.customCommand([
"INFO",
"COMMANDSTATS",
]);

if (typeof info_result === "string") {
expect(info_result).toContain(get_cmdstat);
} else {
throw new Error(
"Unexpected response format from INFO command",
);
}
} finally {
await client_for_testing_az?.close();
}
},
);
});
runBaseTests({
init: async (protocol, configOverrides) => {
const config = getClientConfigurationOption(
cluster.getAddresses(),
protocol,
configOverrides,
);
client = await GlideClient.createClient(config);

const configNew = getClientConfigurationOption(
azCluster.getAddresses(),
protocol,
configOverrides,
);

testsFailed += 1;
azClient = await GlideClient.createClient(configNew);
client = await GlideClient.createClient(config);
return { client, cluster };
return { client, cluster, azClient, azCluster };
},
close: (testSucceeded: boolean) => {
if (testSucceeded) {
Expand Down
Loading
Loading