authorization - gateway - Implement local policy attachment caching f…

services/src/gateway.ts

@@ -16,12 +16,15 @@ import {
 async function run() {
     logger.info('Stitch gateway booting up...');
 
+    const resourceRepository = getResourceRepository();
+
     const {server, dispose} = createStitchGateway({
-        resourceGroups: pollForUpdates(getResourceRepository(), config.resourceUpdateInterval),
+        resourceGroups: pollForUpdates(resourceRepository, config.resourceUpdateInterval),
         tracing: config.enableGraphQLTracing,
         playground: config.enableGraphQLPlayground,
         introspection: config.enableGraphQLIntrospection,
     });
+    await resourceRepository.initializePolicyAttachments();
 
     const app = fastify();
     app.register(fastifyMetrics, {endpoint: '/metrics'});

services/src/modules/resource-repository/composite.ts

@@ -20,4 +20,17 @@ export class CompositeResourceRepository implements ResourceRepository {
     async writePolicyAttachment(): Promise<void> {
         throw new Error('Multiplexed resource repository cannot handle updates');
     }
+
+    public getPolicyAttachment(filename: string): Buffer {
+        for (const repo of this.repositories) {
+            const policyAttachment = repo.getPolicyAttachment(filename);
+            if (policyAttachment) return policyAttachment;
+        }
+
+        throw new Error(`Policy attachment with the filename ${filename} was not found`);
+    }
+
+    public async initializePolicyAttachments() {
+        await Promise.all(this.repositories.map(repo => repo.initializePolicyAttachments()));
+    }
 }

services/src/modules/resource-repository/fs.spec.ts

@@ -0,0 +1,126 @@
+import {promises as fs} from 'fs';
+import * as path from 'path';
+import {FileSystemResourceRepository} from './fs';
+
+let repo: FileSystemResourceRepository;
+const pathToResourcesFile = '/var/stitch/resources.json';
+const policyAttachmentsFolderPath = '/var/stitch/policy-attachments';
+let fsMocks: jest.Mock[];
+
+beforeEach(() => {
+    repo = new FileSystemResourceRepository(pathToResourcesFile, policyAttachmentsFolderPath);
+    fsMocks = [];
+});
+
+afterEach(() => {
+    for (const mock of fsMocks) (mock as any).__restore();
+});
+
+describe('shouldRefreshPolicyAttachment', () => {
+    let shouldRefreshPolicyAttachment: Function;
+
+    beforeEach(() => {
+        shouldRefreshPolicyAttachment = repo['shouldRefreshPolicyAttachment'].bind(repo);
+    });
+
+    it('returns true if the given file does not currently exist in memory', () => {
+        repo['policyAttachmentsRefreshedAt'] = new Date();
+
+        const result = shouldRefreshPolicyAttachment({filename: 'file', updatedAt: minutesAgo(5)});
+        expect(result).toBe(true);
+    });
+
+    it('returns true if this is the first refresh for this process', () => {
+        repo['policyAttachments']['file'] = Buffer.from('content');
+
+        const result = shouldRefreshPolicyAttachment({filename: 'file', updatedAt: minutesAgo(5)});
+        expect(result).toBe(true);
+    });
+
+    it('returns true if the attachment was last updated after the last refresh', () => {
+        repo['policyAttachments']['file'] = Buffer.from('content');
+        repo['policyAttachmentsRefreshedAt'] = minutesAgo(5);
+
+        const result = shouldRefreshPolicyAttachment({filename: 'file', updatedAt: new Date()});
+        expect(result).toBe(true);
+    });
+
+    it('returns false if the attachment was last updated before the last refresh', () => {
+        repo['policyAttachments']['file'] = Buffer.from('content');
+        repo['policyAttachmentsRefreshedAt'] = new Date();
+
+        const result = shouldRefreshPolicyAttachment({filename: 'file', updatedAt: minutesAgo(5)});
+        expect(result).toBe(false);
+    });
+});
+
+describe('refreshPolicyAttachments', () => {
+    let refreshPolicyAttachments: Function;
+
+    beforeEach(() => {
+        refreshPolicyAttachments = repo['refreshPolicyAttachments'].bind(repo);
+    });
+
+    it('refreshes the local copy of all policy attachments that should be refreshed', async () => {
+        repo['policyAttachments'] = {
+            upToDateFile: Buffer.from('up to date'),
+            needsUpdating: Buffer.from('needs updating'),
+        };
+        repo['policyAttachmentsRefreshedAt'] = minutesAgo(5);
+
+        const upToDateFilePath = path.resolve(policyAttachmentsFolderPath, 'upToDateFile');
+        const needsUpdatingPath = path.resolve(policyAttachmentsFolderPath, 'needsUpdating');
+        const newFilePath = path.resolve(policyAttachmentsFolderPath, 'newFile');
+
+        const readdirMock = mockFsFunction('readdir');
+        readdirMock.mockReturnValue(Promise.resolve(['upToDateFile', 'needsUpdating', 'newFile']));
+
+        const statParamBasedReturnValues = {
+            [upToDateFilePath]: {mtime: minutesAgo(10)},
+            [needsUpdatingPath]: {mtime: minutesAgo(2)},
+            [newFilePath]: {mtime: minutesAgo(3)},
+        };
+        const statMock = mockFsFunction('stat');
+        statMock.mockImplementation(filePath => Promise.resolve(statParamBasedReturnValues[filePath]));
+
+        const readFileParamBasedReturnValues = {
+            [needsUpdatingPath]: Buffer.from('needs updating - updated'),
+            [newFilePath]: Buffer.from('new file'),
+        };
+        const readFileMock = mockFsFunction('readFile');
+        readFileMock.mockImplementation(filePath => Promise.resolve(readFileParamBasedReturnValues[filePath]));
+
+        await refreshPolicyAttachments();
+
+        expect(repo['policyAttachments']).toMatchObject({
+            upToDateFile: Buffer.from('up to date'),
+            needsUpdating: Buffer.from('needs updating - updated'),
+            newFile: Buffer.from('new file'),
+        });
+        expect(repo['policyAttachmentsRefreshedAt'].getTime()).toBeGreaterThan(minutesAgo(1).getTime());
+
+        expect(readdirMock).toHaveBeenCalledTimes(1);
+        expect(readdirMock).toHaveBeenCalledWith(policyAttachmentsFolderPath);
+
+        expect(statMock).toHaveBeenCalledTimes(3);
+        expect(statMock).toHaveBeenCalledWith(upToDateFilePath);
+        expect(statMock).toHaveBeenCalledWith(needsUpdatingPath);
+        expect(statMock).toHaveBeenCalledWith(newFilePath);
+
+        expect(readFileMock).toHaveBeenCalledTimes(2);
+        expect(readFileMock).toHaveBeenCalledWith(needsUpdatingPath);
+        expect(readFileMock).toHaveBeenCalledWith(newFilePath);
+    });
+});
+
+function mockFsFunction(functionName: string): jest.Mock {
+    const realFunction = (fs as any)[functionName];
+    const mock: any = jest.fn();
+    mock.__restore = () => ((fs as any)[functionName] = realFunction);
+    (fs as any)[functionName] = mock;
+
+    fsMocks.push(mock);
+    return mock;
+}
+
+const minutesAgo = (m: number) => new Date(Date.now() - m * 60000);

services/src/modules/resource-repository/fs.ts

@@ -2,11 +2,16 @@ import {ResourceGroup, ResourceRepository} from '.';
 import * as envVar from 'env-var';
 import {promises as fs} from 'fs';
 import * as path from 'path';
+import pLimit from 'p-limit';
+import * as config from '../config';
+import logger from '../logger';
 import {FetchLatestResult} from './types';
 
 export class FileSystemResourceRepository implements ResourceRepository {
     protected current?: {mtime: number; rg: ResourceGroup};
     protected policyAttachmentsDirInitialized = false;
+    protected policyAttachments: {[filename: string]: Buffer} = {};
+    protected policyAttachmentsRefreshedAt?: Date;
 
     constructor(protected pathToFile: string, protected policyAttachmentsFolderPath: string) {}
 
@@ -35,6 +40,85 @@ export class FileSystemResourceRepository implements ResourceRepository {
         await fs.writeFile(filePath, content);
     }
 
+    public getPolicyAttachment(filename: string): Buffer {
+        return this.policyAttachments[filename];
+    }
+
+    public async initializePolicyAttachments() {
+        try {
+            await this.refreshPolicyAttachments();
+        } catch (err) {
+            logger.fatal({err}, 'Failed fetching fs policy attachments on startup');
+            throw err;
+        }
+
+        setInterval(async () => {
+            try {
+                await this.refreshPolicyAttachments();
+            } catch (err) {
+                logger.error(
+                    {err},
+                    `Failed refreshing fs policy attachments, last successful refresh was at ${this.policyAttachmentsRefreshedAt}`
+                );
+            }
+        }, config.resourceUpdateInterval);
+    }
+
+    private async refreshPolicyAttachments() {
+        const newRefreshedAt = new Date();
+
+        const allAttachments = await this.getPolicyAttachmentsList();
+        const attachmentsToRefresh = allAttachments
+            .filter(a => this.shouldRefreshPolicyAttachment(a))
+            .map(a => a.filename);
+
+        if (attachmentsToRefresh.length > 0) {
+            const newAttachments = await this.getPolicyAttachments(attachmentsToRefresh);
+            newAttachments.forEach(a => (this.policyAttachments[a.filename] = a.content));
+        }
+
+        this.policyAttachmentsRefreshedAt = newRefreshedAt;
+    }
+
+    private shouldRefreshPolicyAttachment({filename, updatedAt}: {filename: string; updatedAt: Date}) {
+        if (!this.policyAttachments[filename]) return true;
+        if (!this.policyAttachmentsRefreshedAt) return true;
+
+        return updatedAt > this.policyAttachmentsRefreshedAt;
+    }
+
+    private async getPolicyAttachmentsList(): Promise<{filename: string; updatedAt: Date}[]> {
+        const filenames = await fs.readdir(this.policyAttachmentsFolderPath);
+        const limit = pLimit(10);
+
+        return Promise.all(
+            filenames.map(filename => {
+                return limit(async () => {
+                    const filePath = path.resolve(this.policyAttachmentsFolderPath, filename);
+                    const stats = await fs.stat(filePath);
+
+                    const updatedAt = stats.mtime;
+                    return {filename, updatedAt};
+                });
+            })
+        );
+    }
+
+    private async getPolicyAttachments(filenames: string[]): Promise<{filename: string; content: Buffer}[]> {
+        const limit = pLimit(10);
+
+        return Promise.all(
+            filenames.map(filename => {
+                return limit(async () => {
+                    const filePath = path.resolve(this.policyAttachmentsFolderPath, filename);
+                    const content = await fs.readFile(filePath);
+
+                    return {filename, content};
+                });
+            })
+        );
+    }
+
     private async initializePolicyAttachmentsDir() {
         if (this.policyAttachmentsDirInitialized) return;