sasjs/adapter
1
0
Fork 0
mirror of https://github.com/sasjs/adapter.git synced 2025-12-11 01:14:36 +00:00
Code Issues Projects Releases Wiki Activity

feat(job-state): added session state check to doPoll func

Browse Source
This commit is contained in:
Yury Shkoda
2023-09-11 11:17:05 +03:00
parent 0359fcb6be
commit 3a186bc55c
9 changed files with 423 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
jest.config.js
View File

@@ -43,10 +43,10 @@ module.exports = {
// An object that configures minimum threshold enforcement for coverage results
coverageThreshold: {
global: {
statements: 63.61,
branches: 44.72,
functions: 53.94,
lines: 64.07
statements: 64.01,
branches: 45.11,
functions: 54.1,
lines: 64.51
}
},

43
src/SessionManager.ts
View File

@@ -1,4 +1,4 @@
import { Session, Context, SessionVariable } from './types'
import { Session, Context, SessionVariable, SessionState } from './types'
import { NoSessionStateError } from './types/errors'
import { asyncForEach, isUrl } from './utils'
import { prefixMessage } from '@sasjs/utils/error'
@@ -12,6 +12,7 @@ interface ApiErrorResponse {
export class SessionManager {
private loggedErrors: NoSessionStateError[] = []
private sessionStateLinkError = 'Error while getting session state link. '
constructor(
private serverUrl: string,
@@ -28,7 +29,7 @@ export class SessionManager {
private _debug: boolean = false
private printedSessionState = {
printed: false,
state: ''
state: SessionState.NoState
}
public get debug() {
@@ -265,6 +266,18 @@ export class SessionManager {
)
})
// Add response etag to Session object.
createdSession.etag = etag
// Get session state link.
const stateLink = createdSession.links.find((link) => link.rel === 'state')
// Throw error if session state link is not present.
if (!stateLink) throw this.sessionStateLinkError
// Add session state link to Session object.
createdSession.stateUrl = stateLink.href
await this.waitForSession(createdSession, etag, accessToken)
this.sessions.push(createdSession)
@@ -327,32 +340,30 @@ export class SessionManager {
etag: string | null,
accessToken?: string
): Promise<string> {
let { state: sessionState } = session
const { stateUrl } = session
const logger = process.logger || console
let sessionState = session.state
const stateLink = session.links.find((l: any) => l.rel === 'state')
if (
sessionState === 'pending' ||
sessionState === 'running' ||
sessionState === ''
sessionState === SessionState.Pending ||
sessionState === SessionState.Running ||
sessionState === SessionState.NoState
) {
if (stateLink) {
if (stateUrl) {
if (this.debug && !this.printedSessionState.printed) {
logger.info(`Polling: ${this.serverUrl + stateLink.href}`)
logger.info(`Polling: ${this.serverUrl + stateUrl}`)
this.printedSessionState.printed = true
}
const url = `${this.serverUrl}${stateLink.href}?wait=30`
const url = `${this.serverUrl}${stateUrl}?wait=30`
const { result: state, responseStatus: responseStatus } =
await this.getSessionState(url, etag!, accessToken).catch((err) => {
throw prefixMessage(err, 'Error while waiting for session. ')
})
sessionState = state.trim()
sessionState = state.trim() as SessionState
if (this.debug && this.printedSessionState.state !== sessionState) {
logger.info(`Current session state is '${sessionState}'`)
@@ -364,7 +375,7 @@ export class SessionManager {
if (!sessionState) {
const stateError = new NoSessionStateError(
responseStatus,
this.serverUrl + stateLink.href,
this.serverUrl + stateUrl,
session.links.find((l: any) => l.rel === 'log')?.href as string
)
@@ -386,7 +397,7 @@ export class SessionManager {
return sessionState
} else {
throw 'Error while getting session state link. '
throw this.sessionStateLinkError
}
} else {
this.loggedErrors = []
@@ -413,7 +424,7 @@ export class SessionManager {
return await this.requestClient
.get(url, accessToken, 'text/plain', { 'If-None-Match': etag })
.then((res) => ({
result: res.result as string,
result: res.result as SessionState,
responseStatus: res.status
}))
.catch((err) => {

19
src/api/viya/executeOnComputeApi.ts
View File

@@ -170,16 +170,21 @@ export async function executeOnComputeApi(
postedJob,
debug,
authConfig,
pollOptions
pollOptions,
{
session,
sessionManager
}
).catch(async (err) => {
const error = err?.response?.data
const result = /err=[0-9]*,/.exec(error)
const errorCode = '5113'
if (result?.[0]?.slice(4, -1) === errorCode) {
const logCount = 1000000
const sessionLogUrl =
postedJob.links.find((l: any) => l.rel === 'up')!.href + '/log'
const logCount = 1000000
err.log = await fetchLogByChunks(
requestClient,
access_token!,
@@ -187,6 +192,7 @@ export async function executeOnComputeApi(
logCount
)
}
throw prefixMessage(err, 'Error while polling job status. ')
})
@@ -205,12 +211,12 @@ export async function executeOnComputeApi(
let jobResult
let log = ''
const logLink = currentJob.links.find((l) => l.rel === 'log')
if (debug && logLink) {
const logUrl = `${logLink.href}/content`
const logCount = currentJob.logStatistics?.lineCount ?? 1000000
log = await fetchLogByChunks(
requestClient,
access_token!,
@@ -223,9 +229,7 @@ export async function executeOnComputeApi(
throw new ComputeJobExecutionError(currentJob, log)
}
if (!expectWebout) {
return { job: currentJob, log }
}
if (!expectWebout) return { job: currentJob, log }
const resultLink = `/compute/sessions/${executionSessionId}/filerefs/_webout/content`
@@ -236,6 +240,7 @@ export async function executeOnComputeApi(
if (logLink) {
const logUrl = `${logLink.href}/content`
const logCount = currentJob.logStatistics?.lineCount ?? 1000000
log = await fetchLogByChunks(
requestClient,
access_token!,

60
src/api/viya/pollJobState.ts
View File

@@ -3,7 +3,7 @@ import { Job, PollOptions, PollStrategy } from '../..'
import { getTokens } from '../../auth/getTokens'
import { RequestClient } from '../../request/RequestClient'
import { JobStatePollError } from '../../types/errors'
import { Link, WriteStream } from '../../types'
import { Link, WriteStream, SessionState, JobSessionManager } from '../../types'
import { delay, isNode } from '../../utils'
export enum JobState {
@@ -37,6 +37,7 @@ export enum JobState {
* { maxPollCount: 500, pollInterval: 30000 }, // approximately ~50.5 mins (including time to get response (~300ms))
* { maxPollCount: 3400, pollInterval: 60000 } // approximately ~3015 mins (~125 hours) (including time to get response (~300ms))
* ]
* @param jobSessionManager - job session object containing session object and an instance of Session Manager. Job session object is used to periodically (every 10th job state poll) check parent session state.
* @returns - a promise which resolves with a job state
*/
export async function pollJobState(
@@ -44,7 +45,8 @@ export async function pollJobState(
postedJob: Job,
debug: boolean,
authConfig?: AuthConfig,
pollOptions?: PollOptions
pollOptions?: PollOptions,
jobSessionManager?: JobSessionManager
): Promise<JobState> {
const logger = process.logger || console
@@ -127,7 +129,8 @@ export async function pollJobState(
pollOptions,
authConfig,
streamLog,
logFileStream
logFileStream,
jobSessionManager
)
currentState = result.state
@@ -158,7 +161,8 @@ export async function pollJobState(
defaultPollOptions,
authConfig,
streamLog,
logFileStream
logFileStream,
jobSessionManager
)
currentState = result.state
@@ -208,7 +212,21 @@ const needsRetry = (state: string) =>
state === JobState.Pending ||
state === JobState.Unavailable
const doPoll = async (
/**
* Polls job state.
* @param requestClient - the pre-configured HTTP request client.
* @param postedJob - the relative or absolute path to the job.
* @param currentState - current job state.
* @param debug - sets the _debug flag in the job arguments.
* @param pollCount - current poll count.
* @param pollOptions - an object containing maxPollCount, pollInterval, streamLog and logFolderPath.
* @param authConfig - an access token, refresh token, client and secret for an authorized user.
* @param streamLog - indicates if job log should be streamed.
* @param logStream - job log stream.
* @param jobSessionManager - job session object containing session object and an instance of Session Manager. Job session object is used to periodically (every 10th job state poll) check parent session state.
* @returns - a promise which resolves with a job state
*/
export const doPoll = async (
requestClient: RequestClient,
postedJob: Job,
currentState: JobState,
@@ -217,7 +235,8 @@ const doPoll = async (
pollOptions: PollOptions,
authConfig?: AuthConfig,
streamLog?: boolean,
logStream?: WriteStream
logStream?: WriteStream,
jobSessionManager?: JobSessionManager
): Promise<{ state: JobState; pollCount: number }> => {
const { maxPollCount, pollInterval } = pollOptions
const logger = process.logger || console
@@ -229,6 +248,35 @@ const doPoll = async (
let startLogLine = 0
while (needsRetry(state) && pollCount <= maxPollCount) {
// Check parent session state on every 10th job state poll.
if (jobSessionManager && pollCount && pollCount % 10 === 0 && authConfig) {
const { session, sessionManager } = jobSessionManager
const { stateUrl, etag, id: sessionId } = session
const { access_token } = authConfig
const { id: jobId } = postedJob
// Get session state.
const { result: sessionState, responseStatus } = await sessionManager[
'getSessionState'
](stateUrl, etag, access_token).catch((err) => {
// Handle error while getting session state.
throw new JobStatePollError(jobId, err)
})
// Clear parent session and throw an error if session state is not
// 'running' or response status is not 200.
if (sessionState !== SessionState.Running || responseStatus !== 200) {
sessionManager.clearSession(sessionId, access_token)
const sessionError =
sessionState !== SessionState.Running
? `Session state of the job is not 'running'. Session state is '${sessionState}'`
: `Session response status is not 200. Session response status is ${responseStatus}.`
throw new JobStatePollError(jobId, new Error(sessionError))
}
}
state = await getJobState(
requestClient,
postedJob,

10
src/api/viya/spec/executeScript.spec.ts
View File

@@ -7,7 +7,7 @@ import * as uploadTablesModule from '../uploadTables'
import * as getTokensModule from '../../../auth/getTokens'
import * as formatDataModule from '../../../utils/formatDataForRequest'
import * as fetchLogsModule from '../../../utils/fetchLogByChunks'
import { PollOptions } from '../../../types'
import { PollOptions, JobSessionManager } from '../../../types'
import { ComputeJobExecutionError, NotFoundError } from '../../../types/errors'
import { Logger, LogLevel } from '@sasjs/utils/logger'
@@ -308,6 +308,11 @@ describe('executeScript', () => {
})
it('should poll for job completion when waitForResult is true', async () => {
const jobSessionManager: JobSessionManager = {
session: mockSession,
sessionManager: sessionManager
}
await executeOnComputeApi(
requestClient,
sessionManager,
@@ -329,7 +334,8 @@ describe('executeScript', () => {
mockJob,
false,
mockAuthConfig,
defaultPollOptions
defaultPollOptions,
jobSessionManager
)
})

8
src/api/viya/spec/mockResponses.ts
View File

@@ -1,14 +1,16 @@
import { AuthConfig } from '@sasjs/utils/types'
import { Job, Session } from '../../../types'
import { Job, Session, SessionState } from '../../../types'
export const mockSession: Session = {
id: 's35510n',
state: 'idle',
state: SessionState.Idle,
stateUrl: '',
links: [],
attributes: {
sessionInactiveTimeout: 1
},
creationTimeStamp: new Date().valueOf().toString()
creationTimeStamp: new Date().valueOf().toString(),
etag: 'etag-string'
}
export const mockJob: Job = {

224
src/api/viya/spec/pollJobState.spec.ts
View File

@@ -1,17 +1,25 @@
import { Logger, LogLevel } from '@sasjs/utils/logger'
import { RequestClient } from '../../../request/RequestClient'
import { mockAuthConfig, mockJob } from './mockResponses'
import { pollJobState } from '../pollJobState'
import { pollJobState, doPoll, JobState } from '../pollJobState'
import * as getTokensModule from '../../../auth/getTokens'
import * as saveLogModule from '../saveLog'
import * as getFileStreamModule from '../getFileStream'
import * as isNodeModule from '../../../utils/isNode'
import * as delayModule from '../../../utils/delay'
import { PollOptions, PollStrategy } from '../../../types'
import {
PollOptions,
PollStrategy,
SessionState,
JobSessionManager
} from '../../../types'
import { WriteStream } from 'fs'
import { SessionManager } from '../../../SessionManager'
import { JobStatePollError } from '../../../types'
const baseUrl = 'http://localhost'
const requestClient = new (<jest.Mock<RequestClient>>RequestClient)()
const sessionManager = new (<jest.Mock<SessionManager>>SessionManager)()
requestClient['httpClient'].defaults.baseURL = baseUrl
const defaultStreamLog = false
@@ -423,6 +431,218 @@ describe('pollJobState', () => {
})
})
describe('doPoll', () => {
const sessionStateLink = '/compute/sessions/session-id-ses0000/state'
const jobSessionManager: JobSessionManager = {
sessionManager,
session: {
id: ['id', new Date().getTime(), Math.random()].join('-'),
state: SessionState.NoState,
links: [
{
href: sessionStateLink,
method: 'GET',
rel: 'state',
type: 'text/plain',
uri: sessionStateLink
}
],
attributes: {
sessionInactiveTimeout: 900
},
creationTimeStamp: `${new Date(new Date().getTime()).toISOString()}`,
stateUrl: '',
etag: ''
}
}
beforeEach(() => {
setupMocks()
})
it('should check session state on every 10th job state poll', async () => {
const mockedGetSessionState = jest
.spyOn(sessionManager as any, 'getSessionState')
.mockImplementation(() => {
return Promise.resolve({
result: SessionState.Running,
responseStatus: 200
})
})
let getSessionStateCount = 0
jest.spyOn(requestClient, 'get').mockImplementation(() => {
getSessionStateCount++
return Promise.resolve({
result:
getSessionStateCount < 20 ? JobState.Running : JobState.Completed,
etag: 'etag-string',
status: 200
})
})
await doPoll(
requestClient,
mockJob,
JobState.Running,
false,
1,
defaultPollStrategy,
mockAuthConfig,
undefined,
undefined,
jobSessionManager
)
expect(mockedGetSessionState).toHaveBeenCalledTimes(2)
})
it('should handle error while checking session state', async () => {
const sessionStateError = 'Error while getting session state.'
jest
.spyOn(sessionManager as any, 'getSessionState')
.mockImplementation(() => {
return Promise.reject(sessionStateError)
})
jest.spyOn(requestClient, 'get').mockImplementation(() => {
return Promise.resolve({
result: JobState.Running,
etag: 'etag-string',
status: 200
})
})
await expect(
doPoll(
requestClient,
mockJob,
JobState.Running,
false,
1,
defaultPollStrategy,
mockAuthConfig,
undefined,
undefined,
jobSessionManager
)
).rejects.toEqual(
new JobStatePollError(mockJob.id, new Error(sessionStateError))
)
})
it('should throw an error if session is not in running state', async () => {
const filteredSessionStates = Object.values(SessionState).filter(
(state) => state !== SessionState.Running
)
const randomSessionState =
filteredSessionStates[
Math.floor(Math.random() * filteredSessionStates.length)
]
jest
.spyOn(sessionManager as any, 'getSessionState')
.mockImplementation(() => {
return Promise.resolve({
result: randomSessionState,
responseStatus: 200
})
})
jest.spyOn(requestClient, 'get').mockImplementation(() => {
return Promise.resolve({
result: JobState.Running,
etag: 'etag-string',
status: 200
})
})
const mockedClearSession = jest
.spyOn(sessionManager, 'clearSession')
.mockImplementation(() => Promise.resolve())
await expect(
doPoll(
requestClient,
mockJob,
JobState.Running,
false,
1,
defaultPollStrategy,
mockAuthConfig,
undefined,
undefined,
jobSessionManager
)
).rejects.toEqual(
new JobStatePollError(
mockJob.id,
new Error(
`Session state of the job is not 'running'. Session state is '${randomSessionState}'`
)
)
)
expect(mockedClearSession).toHaveBeenCalledWith(
jobSessionManager.session.id,
mockAuthConfig.access_token
)
})
it('should handle throw an error if response status of session state is not 200', async () => {
const sessionStateResponseStatus = 500
jest
.spyOn(sessionManager as any, 'getSessionState')
.mockImplementation(() => {
return Promise.resolve({
result: SessionState.Running,
responseStatus: sessionStateResponseStatus
})
})
jest.spyOn(requestClient, 'get').mockImplementation(() => {
return Promise.resolve({
result: JobState.Running,
etag: 'etag-string',
status: 200
})
})
const mockedClearSession = jest
.spyOn(sessionManager, 'clearSession')
.mockImplementation(() => Promise.resolve())
await expect(
doPoll(
requestClient,
mockJob,
JobState.Running,
false,
1,
defaultPollStrategy,
mockAuthConfig,
undefined,
undefined,
jobSessionManager
)
).rejects.toEqual(
new JobStatePollError(
mockJob.id,
new Error(
`Session response status is not 200. Session response status is ${sessionStateResponseStatus}.`
)
)
)
expect(mockedClearSession).toHaveBeenCalledWith(
jobSessionManager.session.id,
mockAuthConfig.access_token
)
})
})
const setupMocks = () => {
jest.restoreAllMocks()
jest.mock('../../../request/RequestClient')

83
src/test/SessionManager.spec.ts
View File

@@ -3,7 +3,7 @@ import { RequestClient } from '../request/RequestClient'
import * as dotenv from 'dotenv'
import axios from 'axios'
import { Logger, LogLevel } from '@sasjs/utils/logger'
import { Session, Context } from '../types'
import { Session, SessionState, Context } from '../types'
jest.mock('axios')
const mockedAxios = axios as jest.Mocked<typeof axios>
@@ -11,21 +11,34 @@ const requestClient = new (<jest.Mock<RequestClient>>RequestClient)()
describe('SessionManager', () => {
dotenv.config()
process.env.SERVER_URL = 'https://server.com'
const sessionManager = new SessionManager(
process.env.SERVER_URL as string,
process.env.DEFAULT_COMPUTE_CONTEXT as string,
requestClient
)
const sessionStateLink = '/compute/sessions/session-id-ses0000/state'
const sessionEtag = 'etag-string'
const getMockSession = () => ({
const getMockSession = (): Session => ({
id: ['id', new Date().getTime(), Math.random()].join('-'),
state: '',
links: [{ rel: 'state', href: '', uri: '', type: '', method: 'GET' }],
state: SessionState.NoState,
links: [
{
href: sessionStateLink,
method: 'GET',
rel: 'state',
type: 'text/plain',
uri: sessionStateLink
}
],
attributes: {
sessionInactiveTimeout: 900
},
creationTimeStamp: `${new Date(new Date().getTime()).toISOString()}`
creationTimeStamp: `${new Date(new Date().getTime()).toISOString()}`,
stateUrl: sessionStateLink,
etag: sessionEtag
})
afterEach(() => {
@@ -89,19 +102,21 @@ describe('SessionManager', () => {
describe('waitForSession', () => {
const session: Session = {
id: 'id',
state: '',
state: SessionState.NoState,
links: [{ rel: 'state', href: '', uri: '', type: '', method: 'GET' }],
attributes: {
sessionInactiveTimeout: 0
},
creationTimeStamp: ''
creationTimeStamp: '',
stateUrl: sessionStateLink,
etag: sessionEtag
}
beforeEach(() => {
;(process as any).logger = new Logger(LogLevel.Off)
})
it('should reject with NoSessionStateError if SAS server did not provide session state', async () => {
it('should log http response code and session state if SAS server did not provide session state', async () => {
let requestAttempt = 0
const requestAttemptLimit = 10
const sessionState = 'idle'
@@ -124,15 +139,17 @@ describe('SessionManager', () => {
sessionManager['waitForSession'](session, null, 'access_token')
).resolves.toEqual(sessionState)
const sessionStateUrl = process.env.SERVER_URL + session.stateUrl
expect(mockedAxios.get).toHaveBeenCalledTimes(requestAttemptLimit)
expect((process as any).logger.info).toHaveBeenCalledTimes(3)
expect((process as any).logger.info).toHaveBeenNthCalledWith(
1,
`Polling: ${process.env.SERVER_URL}`
`Polling: ${sessionStateUrl}`
)
expect((process as any).logger.info).toHaveBeenNthCalledWith(
2,
`Could not get session state. Server responded with 304 whilst checking state: ${process.env.SERVER_URL}`
`Could not get session state. Server responded with 304 whilst checking state: ${sessionStateUrl}`
)
expect((process as any).logger.info).toHaveBeenNthCalledWith(
3,
@@ -142,7 +159,7 @@ describe('SessionManager', () => {
it('should throw an error if there is no session link', async () => {
const customSession = JSON.parse(JSON.stringify(session))
customSession.links = []
customSession.stateUrl = ''
mockedAxios.get.mockImplementation(() =>
Promise.resolve({ data: customSession.state, status: 200 })
@@ -156,6 +173,7 @@ describe('SessionManager', () => {
it('should throw an error if could not get session state', async () => {
const gettingSessionStatus = 500
const sessionStatusError = `Getting session status timed out after 60 seconds. Request failed with status code ${gettingSessionStatus}`
const sessionStateUrl = process.env.SERVER_URL + session.stateUrl
mockedAxios.get.mockImplementation(() =>
Promise.reject({
@@ -168,7 +186,7 @@ describe('SessionManager', () => {
})
)
const expectedError = `Error while waiting for session. Error while getting session state. GET request to ${process.env.SERVER_URL}?wait=30 failed with status code ${gettingSessionStatus}. ${sessionStatusError}`
const expectedError = `Error while waiting for session. Error while getting session state. GET request to ${sessionStateUrl}?wait=30 failed with status code ${gettingSessionStatus}. ${sessionStatusError}`
await expect(
sessionManager['waitForSession'](session, null, 'access_token')
@@ -427,4 +445,45 @@ describe('SessionManager', () => {
)
})
})
describe('createAndWaitForSession', () => {
it('should create session with etag and stateUrl', async () => {
const etag = sessionEtag
const customSession: any = getMockSession()
delete customSession.etag
delete customSession.stateUrl
jest.spyOn(requestClient, 'post').mockImplementation(() =>
Promise.resolve({
result: customSession,
etag
})
)
jest
.spyOn(sessionManager as any, 'setCurrentContext')
.mockImplementation(() => Promise.resolve())
sessionManager['currentContext'] = {
name: 'context name',
id: 'string',
createdBy: 'string',
version: 1
}
jest
.spyOn(sessionManager as any, 'getSessionState')
.mockImplementation(() =>
Promise.resolve({ result: SessionState.Idle, responseStatus: 200 })
)
const expectedSession = await sessionManager['createAndWaitForSession']()
expect(customSession.id).toEqual(expectedSession.id)
expect(
customSession.links.find((l: any) => l.rel === 'state').href
).toEqual(expectedSession.stateUrl)
expect(expectedSession.etag).toEqual(etag)
})
})
})

21
src/types/Session.ts
View File

@@ -1,15 +1,34 @@
import { Link } from './Link'
import { SessionManager } from '../SessionManager'
export enum SessionState {
Completed = 'completed',
Running = 'running',
Pending = 'pending',
Idle = 'idle',
Unavailable = 'unavailable',
NoState = '',
Failed = 'failed',
Error = 'error'
}
export interface Session {
id: string
state: string
state: SessionState
stateUrl: string
links: Link[]
attributes: {
sessionInactiveTimeout: number
}
creationTimeStamp: string
etag: string
}
export interface SessionVariable {
value: string
}
export interface JobSessionManager {
session: Session
sessionManager: SessionManager
}