* Extracts necessary information in the worker from pg_stat_activity. * pg_stat_activity has a PID, we also get the opid, global_pid, lock info, * running information, and join it with pg_locks to get any tables being * accessed by this query. * Each activity is parsed into a SingleWorkerActivity struct to be processed * later in the WriteOneActivityToDocument function */
| 467 | * later in the WriteOneActivityToDocument function |
| 468 | */ |
| 469 | static List * |
| 470 | WorkerGetBaseActivities() |
| 471 | { |
| 472 | /* Now that we're on the worker node, first get the query of operations */ |
| 473 | |
| 474 | /* This is similar to the original currentOp query in SQL (Pre 1.9 ) |
| 475 | * Add new fields to the end of the selectors only */ |
| 476 | StringInfo queryInfo = makeStringInfo(); |
| 477 | appendStringInfoString(queryInfo, |
| 478 | " SELECT " |
| 479 | " pa.query AS query, " |
| 480 | " pa.pid::bigint AS stat_pid, " |
| 481 | " pa.state AS state, " |
| 482 | " pa.global_pid::bigint AS global_pid, " |
| 483 | " collectionRawName AS mongo_collection_raw, " |
| 484 | " EXTRACT(epoch FROM pa.query_start)::bigint AS query_start, " |
| 485 | " EXTRACT(epoch FROM now() - pa.query_start)::bigint AS secs_running, " |
| 486 | " pa.wait_event_type AS wait_event_type, " |
| 487 | " pa.global_pid || ':' || (EXTRACT(epoch FROM pa.query_start) * 1000000)::numeric(20,0) AS op_id, " |
| 488 | " EXTRACT(epoch FROM now() - pa.state_change)::bigint AS state_change_since, " |
| 489 | " pa.groupid AS shard_id, " |
| 490 | " pa.backend_type AS backend_type, " |
| 491 | " pa.leader_pid::bigint AS leaderPid " |
| 492 | " FROM ("); |
| 493 | |
| 494 | appendStringInfoString(queryInfo, DistributedOperationsQuery); |
| 495 | |
| 496 | /* To get the collections associated with the command join with locks */ |
| 497 | appendStringInfoString(queryInfo, ") pa LEFT JOIN lateral " |
| 498 | " ( " |
| 499 | " SELECT c.relname::text AS collectionRawName " |
| 500 | " FROM pg_locks pl " |
| 501 | " JOIN pg_class c ON (pl.relation = c.oid) " |
| 502 | " JOIN pg_namespace nsp ON (c.relnamespace = nsp.oid) "); |
| 503 | |
| 504 | appendStringInfo(queryInfo, |
| 505 | " WHERE nsp.nspname = '%s' AND c.relkind = 'r' AND pl.pid = pa.pid LIMIT 1", |
| 506 | ApiDataSchemaName); |
| 507 | |
| 508 | appendStringInfo(queryInfo, |
| 509 | " ) e2 ON true WHERE (NOT query LIKE '%%%s.current_op%%') ", |
| 510 | ApiToApiInternalSchemaName); |
| 511 | |
| 512 | if (CurrentOpApplicationName != NULL && |
| 513 | strlen(CurrentOpApplicationName) > 0) |
| 514 | { |
| 515 | appendStringInfo(queryInfo, |
| 516 | " AND (application_name = '%s' OR application_name LIKE '%%%s'", |
| 517 | CurrentOpApplicationName, GetExtensionApplicationName()); |
| 518 | |
| 519 | if (DistributedApplicationNamePrefix != NULL) |
| 520 | { |
| 521 | appendStringInfo(queryInfo, |
| 522 | " OR (application_name LIKE '%s%%' AND worker_query AND state != 'idle')", |
| 523 | DistributedApplicationNamePrefix); |
| 524 | } |
| 525 | |
| 526 | appendStringInfoString(queryInfo, ") "); |
no test coverage detected