Skip to content

Commit 3d56225

Browse files
authored
feat: add offline mode, fix in-process connection edge cases (#708)
Signed-off-by: Michael Beemer <[email protected]>
1 parent 35ae705 commit 3d56225

15 files changed

+522
-245
lines changed

libs/providers/flagd/README.md

+15-3
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
# Server-Side flagd Provider
22

3-
Flagd is a simple daemon for evaluating feature flags.
4-
It is designed to conform to OpenFeature schema for flag definitions.
3+
This provider is designed to use flagd's [evaluation protocol](https://github.com/open-feature/schemas/blob/main/protobuf/schema/v1/schema.proto), or locally evaluate flags defined in a flagd [flag definition](https://github.com/open-feature/schemas/blob/main/json/flagd-definitions.json).
54
This repository and package provides the client code for interacting with it via the OpenFeature server-side JavaScript SDK.
65

76
## Installation
@@ -45,7 +44,7 @@ Below are examples of usage patterns.
4544

4645
This is the default mode of operation of the provider.
4746
In this mode, FlagdProvider communicates with flagd via the gRPC protocol.
48-
Flag evaluations take place remotely at the connected [flagd](https://flagd.dev/) instance.
47+
Flag evaluations take place remotely on the connected [flagd](https://flagd.dev/) instance.
4948

5049
```ts
5150
OpenFeature.setProvider(new FlagdProvider())
@@ -74,6 +73,19 @@ Flag configurations for evaluation are obtained via gRPC protocol using [sync pr
7473

7574
In the above example, the provider expects a flag sync service implementation to be available at `localhost:8013` (default host and port).
7675

76+
In-process resolver can also work in an offline mode.
77+
To enable this mode, you should provide a valid flag configuration file with the option `offlineFlagSourcePath`.
78+
79+
```
80+
OpenFeature.setProvider(new FlagdProvider({
81+
resolverType: 'in-process',
82+
offlineFlagSourcePath: './flags.json',
83+
}))
84+
```
85+
86+
Offline mode uses `fs.watchFile` and polls every 5 seconds for changes to the file.
87+
This mode is useful for local development, test cases, and for offline applications.
88+
7789
### Supported Events
7890

7991
The flagd provider emits `PROVIDER_READY`, `PROVIDER_ERROR` and `PROVIDER_CONFIGURATION_CHANGED` events.

libs/providers/flagd/jest.config.ts

+8-8
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,16 @@
22
export default {
33
displayName: 'providers-flagd',
44
preset: '../../../jest.preset.js',
5-
globals: {
6-
'ts-jest': {
7-
tsconfig: '<rootDir>/tsconfig.spec.json',
8-
},
9-
},
105
transform: {
11-
'^.+\\.[tj]s$': 'ts-jest',
6+
'^.+\\.[tj]s$': [
7+
'ts-jest',
8+
{
9+
tsconfig: '<rootDir>/tsconfig.spec.json',
10+
},
11+
],
1212
},
1313
moduleFileExtensions: ['ts', 'js', 'html'],
1414
// ignore e2e path
15-
testPathIgnorePatterns: ["/e2e/"],
16-
coverageDirectory: '../../../coverage/libs/providers/flagd'
15+
testPathIgnorePatterns: ['/e2e/'],
16+
coverageDirectory: '../../../coverage/libs/providers/flagd',
1717
};

libs/providers/flagd/src/e2e/step-definitions/flagd.spec.ts

+8-6
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { OpenFeature, ProviderEvents } from '@openfeature/server-sdk';
1+
import { OpenFeature, ProviderEvents, EventDetails } from '@openfeature/server-sdk';
22
import { defineFeature, loadFeature } from 'jest-cucumber';
33

44
// load the feature file.
@@ -38,22 +38,24 @@ defineFeature(feature, (test) => {
3838

3939
test('Flag change event', ({ given, when, and, then }) => {
4040
let ran = false;
41+
let eventDetails: EventDetails<ProviderEvents> | undefined;
4142

4243
aFlagProviderIsSet(given);
4344
when('a PROVIDER_CONFIGURATION_CHANGED handler is added', () => {
44-
client.addHandler(ProviderEvents.ConfigurationChanged, async () => {
45+
client.addHandler(ProviderEvents.ConfigurationChanged, async (details) => {
4546
ran = true;
47+
eventDetails = details;
4648
});
4749
});
4850
and(/^a flag with key "(.*)" is modified$/, async () => {
49-
// this happens every 1s in the associated container, so wait 2s
50-
await new Promise((resolve) => setTimeout(resolve, 2000));
51+
// this happens every 1s in the associated container, so wait 3s
52+
await new Promise((resolve) => setTimeout(resolve, 3000));
5153
});
5254
then('the PROVIDER_CONFIGURATION_CHANGED handler must run', () => {
5355
expect(ran).toBeTruthy();
5456
});
55-
and(/^the event details must indicate "(.*)" was altered$/, () => {
56-
// not supported
57+
and(/^the event details must indicate "(.*)" was altered$/, (flagName) => {
58+
expect(eventDetails?.flagsChanged).toContain(flagName);
5759
});
5860
});
5961

libs/providers/flagd/src/lib/configuration.ts

+6
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,12 @@ export interface Config {
4444
*/
4545
resolverType?: ResolverType;
4646

47+
/**
48+
* File source of flags to be used by offline mode.
49+
* Setting this enables the offline mode of the in-process provider.
50+
*/
51+
offlineFlagSourcePath?: string;
52+
4753
/**
4854
* Selector to be used with flag sync gRPC contract.
4955
*/

libs/providers/flagd/src/lib/flagd-provider.ts

+4-3
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,8 @@ export class FlagdProvider implements Provider {
6262
})
6363
.catch((err) => {
6464
this._status = ProviderStatus.ERROR;
65-
this.logger?.error(`${this.metadata.name}: error during initialization: ${err.message}, ${err.stack}`);
65+
this.logger?.error(`${this.metadata.name}: error during initialization: ${err.message}`);
66+
this.logger?.debug(err);
6667
throw err;
6768
});
6869
}
@@ -127,9 +128,9 @@ export class FlagdProvider implements Provider {
127128
this._events.emit(ProviderEvents.Ready);
128129
}
129130

130-
private handleError(): void {
131+
private handleError(message: string): void {
131132
this._status = ProviderStatus.ERROR;
132-
this._events.emit(ProviderEvents.Error);
133+
this._events.emit(ProviderEvents.Error, { message });
133134
}
134135

135136
private handleChanged(flagsChanged: string[]): void {

libs/providers/flagd/src/lib/service/grpc/grpc-service.ts

+13-11
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,7 @@ export class GRPCService implements Service {
9797
connect(
9898
reconnectCallback: () => void,
9999
changedCallback: (flagsChanged: string[]) => void,
100-
disconnectCallback: () => void,
100+
disconnectCallback: (message: string) => void,
101101
): Promise<void> {
102102
return new Promise((resolve, reject) =>
103103
this.listen(reconnectCallback, changedCallback, disconnectCallback, resolve, reject),
@@ -148,7 +148,7 @@ export class GRPCService implements Service {
148148
private listen(
149149
reconnectCallback: () => void,
150150
changedCallback: (flagsChanged: string[]) => void,
151-
disconnectCallback: () => void,
151+
disconnectCallback: (message: string) => void,
152152
resolveConnect?: () => void,
153153
rejectConnect?: (reason: Error) => void,
154154
) {
@@ -185,12 +185,14 @@ export class GRPCService implements Service {
185185
if (data && typeof data === 'object' && 'flags' in data && data?.['flags']) {
186186
const flagChangeMessage = data as FlagChangeMessage;
187187
const flagsChanged: string[] = Object.keys(flagChangeMessage.flags || []);
188-
// remove each changed key from cache
189-
flagsChanged.forEach((key) => {
190-
if (this._cache?.delete(key)) {
191-
this.logger?.debug(`${FlagdProvider.name}: evicted key: ${key} from cache.`);
192-
}
193-
});
188+
if (this._cacheEnabled) {
189+
// remove each changed key from cache
190+
flagsChanged.forEach((key) => {
191+
if (this._cache?.delete(key)) {
192+
this.logger?.debug(`${FlagdProvider.name}: evicted key: ${key} from cache.`);
193+
}
194+
});
195+
}
194196
changedCallback(flagsChanged);
195197
}
196198
}
@@ -199,7 +201,7 @@ export class GRPCService implements Service {
199201
private reconnect(
200202
reconnectCallback: () => void,
201203
changedCallback: (flagsChanged: string[]) => void,
202-
disconnectCallback: () => void,
204+
disconnectCallback: (message: string) => void,
203205
) {
204206
const channel = this._client.getChannel();
205207
channel.watchConnectivityState(channel.getConnectivityState(true), Infinity, () => {
@@ -210,9 +212,9 @@ export class GRPCService implements Service {
210212
private handleError(
211213
reconnectCallback: () => void,
212214
changedCallback: (flagsChanged: string[]) => void,
213-
disconnectCallback: () => void,
215+
disconnectCallback: (message: string) => void,
214216
) {
215-
disconnectCallback();
217+
disconnectCallback('streaming connection error, will attempt reconnect...');
216218
this.logger?.error(`${FlagdProvider.name}: streaming connection error, will attempt reconnect...`);
217219
this._cache?.clear();
218220
this.reconnect(reconnectCallback, changedCallback, disconnectCallback);

libs/providers/flagd/src/lib/service/in-process/data-fetch.ts

+25-3
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,34 @@
22
* Contract of in-process resolver's data fetcher
33
*/
44
export interface DataFetch {
5+
/**
6+
* Connects the data fetcher
7+
*/
58
connect(
6-
dataFillCallback: (flags: string) => void,
9+
/**
10+
* Callback that runs when data is received from the source
11+
* @param flags The flags from the source
12+
* @returns The flags that have changed
13+
*/
14+
dataCallback: (flags: string) => string[],
15+
/**
16+
* Callback that runs when the connection is re-established
17+
*/
718
reconnectCallback: () => void,
19+
/**
20+
* Callback that runs when flags have changed
21+
* @param flagsChanged The flags that have changed
22+
*/
823
changedCallback: (flagsChanged: string[]) => void,
9-
disconnectCallback: () => void,
24+
/**
25+
* Callback that runs when the connection is disconnected
26+
* @param message The reason for the disconnection
27+
*/
28+
disconnectCallback: (message: string) => void,
1029
): Promise<void>;
1130

12-
disconnect(): void;
31+
/**
32+
* Disconnects the data fetcher
33+
*/
34+
disconnect(): Promise<void>;
1335
}
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
import fs from 'fs';
2+
import { FileFetch } from './file-fetch';
3+
import { FlagdCore } from '@openfeature/flagd-core';
4+
import { Logger } from '@openfeature/core';
5+
6+
jest.mock('fs', () => ({
7+
...jest.requireActual('fs'),
8+
promises: {
9+
readFile: jest.fn(),
10+
},
11+
}));
12+
13+
const dataFillCallbackMock = jest.fn();
14+
const reconnectCallbackMock = jest.fn();
15+
const changedCallbackMock = jest.fn();
16+
const loggerMock: Logger = {
17+
debug: jest.fn(),
18+
info: jest.fn(),
19+
warn: jest.fn(),
20+
error: jest.fn(),
21+
};
22+
23+
describe('FileFetch', () => {
24+
let flagdCore: FlagdCore;
25+
let fileFetch: FileFetch;
26+
let dataFillCallback: (flags: string) => string[];
27+
28+
beforeEach(() => {
29+
flagdCore = new FlagdCore();
30+
fileFetch = new FileFetch('./flags.json', loggerMock);
31+
dataFillCallback = (flags: string) => {
32+
return flagdCore.setConfigurations(flags);
33+
};
34+
});
35+
36+
afterEach(() => {
37+
jest.resetAllMocks();
38+
});
39+
40+
it('should connect to the file and setup the watcher', async () => {
41+
const flags = '{"flags":{"flag":{"state":"ENABLED","variants":{"on":true,"off":false},"defaultVariant":"off"}}}';
42+
mockReadFile(flags);
43+
const watchMock = jest.fn();
44+
45+
fs.watchFile = watchMock as jest.MockedFunction<typeof fs.watchFile>;
46+
47+
await fileFetch.connect(dataFillCallbackMock, reconnectCallbackMock, changedCallbackMock);
48+
49+
expect(dataFillCallbackMock).toHaveBeenCalledWith(flags);
50+
expect(watchMock).toHaveBeenCalledWith('./flags.json', expect.any(Function));
51+
});
52+
53+
it('should throw because of invalid json', async () => {
54+
const flags = 'this is not JSON';
55+
mockReadFile(flags);
56+
const watchSpy = jest.spyOn(fs, 'watchFile');
57+
58+
await expect(fileFetch.connect(dataFillCallback, reconnectCallbackMock, changedCallbackMock)).rejects.toThrow();
59+
expect(watchSpy).not.toHaveBeenCalled();
60+
});
61+
62+
it('should throw an error if the file is not found', async () => {
63+
const mockReadFile = fs.promises.readFile as jest.MockedFunction<typeof fs.promises.readFile>;
64+
mockReadFile.mockRejectedValue({ code: 'ENOENT' });
65+
66+
await expect(fileFetch.connect(dataFillCallbackMock, reconnectCallbackMock, changedCallbackMock)).rejects.toThrow(
67+
'File not found: ./flags.json',
68+
);
69+
});
70+
71+
it('should throw an error if the file is not accessible', async () => {
72+
const mockReadFile = fs.promises.readFile as jest.MockedFunction<typeof fs.promises.readFile>;
73+
mockReadFile.mockRejectedValue({ code: 'EACCES' });
74+
75+
await expect(fileFetch.connect(dataFillCallbackMock, reconnectCallbackMock, changedCallbackMock)).rejects.toThrow(
76+
'File not accessible: ./flags.json',
77+
);
78+
});
79+
80+
it('should close the watcher on disconnect', async () => {
81+
const watchSpy = jest.spyOn(fs, 'watchFile');
82+
const unwatchSpy = jest.spyOn(fs, 'unwatchFile');
83+
84+
await fileFetch.connect(dataFillCallbackMock, reconnectCallbackMock, changedCallbackMock);
85+
await fileFetch.disconnect();
86+
87+
expect(watchSpy).toHaveBeenCalled();
88+
expect(unwatchSpy).toHaveBeenCalledWith('./flags.json');
89+
});
90+
91+
describe('on file change', () => {
92+
it('should call changedCallback with the changed flags', async () => {
93+
const flags = '{"flags":{"flag":{"state":"ENABLED","variants":{"on":true,"off":false},"defaultVariant":"off"}}}';
94+
const changedFlags =
95+
'{"flags":{"flag":{"state":"ENABLED","variants":{"on":true,"off":false},"defaultVariant":"on"}}}';
96+
const mockReadFile = fs.promises.readFile as jest.MockedFunction<typeof fs.promises.readFile>;
97+
mockReadFile.mockResolvedValueOnce(flags);
98+
const watchMock = jest.fn();
99+
fs.watchFile = watchMock as jest.MockedFunction<typeof fs.watchFile>;
100+
101+
await fileFetch.connect(dataFillCallback, reconnectCallbackMock, changedCallbackMock);
102+
mockReadFile.mockResolvedValueOnce(changedFlags);
103+
// Manually call the callback that is passed to fs.watchFile;
104+
await watchMock.mock.calls[0][1]();
105+
106+
expect(changedCallbackMock).toHaveBeenCalledWith(['flag']);
107+
});
108+
109+
it('should call skip changedCallback because no flag has changed', async () => {
110+
const flags = '{"flags":{"flag":{"state":"ENABLED","variants":{"on":true,"off":false},"defaultVariant":"off"}}}';
111+
const mockReadFile = fs.promises.readFile as jest.MockedFunction<typeof fs.promises.readFile>;
112+
mockReadFile.mockResolvedValue(flags);
113+
const watchMock = jest.fn();
114+
fs.watchFile = watchMock as jest.MockedFunction<typeof fs.watchFile>;
115+
116+
await fileFetch.connect(dataFillCallback, reconnectCallbackMock, changedCallbackMock);
117+
// Manually call the callback that is passed to fs.watchFile;
118+
await watchMock.mock.calls[0][1]();
119+
120+
expect(changedCallbackMock).not.toHaveBeenCalled();
121+
});
122+
123+
it('should log an error if the file could not be read', async () => {
124+
const flags = '{"flags":{"flag":{"state":"ENABLED","variants":{"on":true,"off":false},"defaultVariant":"off"}}}';
125+
const mockReadFile = fs.promises.readFile as jest.MockedFunction<typeof fs.promises.readFile>;
126+
mockReadFile.mockResolvedValue(flags);
127+
const watchMock = jest.fn();
128+
fs.watchFile = watchMock as jest.MockedFunction<typeof fs.watchFile>;
129+
130+
await fileFetch.connect(dataFillCallback, reconnectCallbackMock, changedCallbackMock);
131+
mockReadFile.mockRejectedValueOnce(new Error('Error reading file'));
132+
// Manually call the callback that is passed to fs.watchFile;
133+
await watchMock.mock.calls[0][1]();
134+
135+
expect(changedCallbackMock).not.toHaveBeenCalled();
136+
expect(loggerMock.error).toHaveBeenCalled();
137+
});
138+
});
139+
});
140+
141+
// Helper function to mock fs.promise.readFile
142+
function mockReadFile(flags: string): void {
143+
const mockReadFile = fs.promises.readFile as jest.MockedFunction<typeof fs.promises.readFile>;
144+
mockReadFile.mockResolvedValue(flags);
145+
}

0 commit comments

Comments
 (0)