fix container specific logging

[jrhizor]

I wasn't rigorous when reviewing #7268. The output looked roughly correct. Some lines had source labels, others had destination labels, but they didn't represent the full logs we were supposed to capture.

Example this PR

2021-11-04 22:47:14 INFO () WorkerRun(call):49 - Executing worker wrapper. Airbyte version: dev
2021-11-04 22:47:15 INFO () TemporalAttemptExecution(get):116 - Executing worker wrapper. Airbyte version: dev
2021-11-04 22:47:15 WARN () Databases(createPostgresDatabaseWithRetry):41 - Waiting for database to become available...
2021-11-04 22:47:15 INFO () JobsDatabaseInstance(lambda$static$2):25 - Testing if jobs database is ready...
2021-11-04 22:47:15 INFO () Databases(createPostgresDatabaseWithRetry):58 - Database available!
2021-11-04 22:47:16 WARN () JsonMetaSchema(newValidator):338 - Unknown keyword example - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
2021-11-04 22:47:16 WARN () JsonMetaSchema(newValidator):338 - Unknown keyword existingJavaType - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
2021-11-04 22:47:16 INFO () DefaultReplicationWorker(run):82 - start sync worker. job id: 13 attempt id: 0
2021-11-04 22:47:16 INFO () DefaultReplicationWorker(run):91 - configured sync modes: {public._airbyte_raw_asdfads_airbyte_raw_pokemon=full_refresh - append, public._airbyte_raw_asasdfasdfsdasapokemon=full_refresh - append}
2021-11-04 22:47:16 INFO () DefaultAirbyteDestination(start):64 - Running destination...
2021-11-04 22:47:16 INFO () LineGobbler(voidCall):82 - Checking if airbyte/destination-local-json:0.2.8 exists...
2021-11-04 22:47:16 INFO () LineGobbler(voidCall):82 - airbyte/destination-local-json:0.2.8 was found locally.
2021-11-04 22:47:16 INFO () DockerProcessFactory(create):127 - Preparing command: docker run --rm --init -i -v airbyte_workspace:/data -v /tmp/airbyte_local:/local -w /data/13/0 --network host --log-driver none airbyte/destination-local-json:0.2.8 write --config destination_config.json --catalog destination_catalog.json
2021-11-04 22:47:16 INFO () LineGobbler(voidCall):82 - Checking if airbyte/source-postgres:0.3.13 exists...
2021-11-04 22:47:17 INFO () LineGobbler(voidCall):82 - airbyte/source-postgres:0.3.13 was found locally.
2021-11-04 22:47:17 INFO () DockerProcessFactory(create):127 - Preparing command: docker run --rm --init -i -v airbyte_workspace:/data -v /tmp/airbyte_local:/local -w /data/13/0 --network host --log-driver none airbyte/source-postgres:0.3.13 read --config source_config.json --catalog source_catalog.json
2021-11-04 22:47:17 INFO () DefaultReplicationWorker(lambda$getDestinationOutputRunnable$3):226 - Destination output thread started.
2021-11-04 22:47:17 INFO () DefaultReplicationWorker(run):119 - Waiting for source thread to join.
2021-11-04 22:47:17 INFO () DefaultReplicationWorker(lambda$getReplicationRunnable$2):190 - Replication thread started.
�[34msource�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.s.p.PostgresSource(main):262 - {} - starting source: class io.airbyte.integrations.source.postgres.PostgresSource
�[35mdestination�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):96 - {} - Running integration: io.airbyte.integrations.destination.local_json.LocalJsonDestination
�[35mdestination�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationCliParser(parseOptions):135 - {} - integration args: {catalog=destination_catalog.json, write=null, config=destination_config.json}
�[35mdestination�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):100 - {} - Command: WRITE
�[35mdestination�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):101 - {} - Integration config: IntegrationConfig{command=WRITE, configPath='destination_config.json', catalogPath='destination_catalog.json', statePath='null'}
�[35mdestination�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[33mWARN�[m c.n.s.JsonMetaSchema(newValidator):338 - {} - Unknown keyword examples - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
�[34msource�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):76 - {} - Running integration: io.airbyte.integrations.base.ssh.SshWrappedSource
�[34msource�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationCliParser(parseOptions):115 - {} - integration args: {read=null, catalog=source_catalog.json, config=source_config.json}
�[34msource�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):80 - {} - Command: READ
�[34msource�[0m - 2021-11-04 22:47:20 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:20 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):81 - {} - Integration config: IntegrationConfig{command=READ, configPath='source_config.json', catalogPath='source_catalog.json', statePath='null'}
�[34msource�[0m - 2021-11-04 22:47:21 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:21 �[33mWARN�[m c.n.s.JsonMetaSchema(newValidator):338 - {} - Unknown keyword order - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
�[34msource�[0m - 2021-11-04 22:47:21 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:21 �[33mWARN�[m c.n.s.JsonMetaSchema(newValidator):338 - {} - Unknown keyword examples - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
�[34msource�[0m - 2021-11-04 22:47:21 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:21 �[33mWARN�[m c.n.s.JsonMetaSchema(newValidator):338 - {} - Unknown keyword airbyte_secret - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
�[34msource�[0m - 2021-11-04 22:47:21 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:21 �[33mWARN�[m c.n.s.JsonMetaSchema(newValidator):338 - {} - Unknown keyword multiline - you should define your own Meta Schema. If the keyword is irrelevant for validation, just use a NonValidationKeyword
�[35mdestination�[0m - 2021-11-04 22:47:21 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:21 �[32mINFO�[m i.a.i.d.l.LocalJsonDestination$JsonConsumer(<init>):151 - {} - initializing consumer.
�[34msource�[0m - 2021-11-04 22:47:21 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:21 �[32mINFO�[m i.a.i.b.s.SshTunnel(getInstance):170 - {} - Starting connection with method: NO_TUNNEL
�[34msource�[0m - 2021-11-04 22:47:23 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:23 �[32mINFO�[m i.a.i.s.p.PostgresSource(isCdc):212 - {} - using CDC: false
�[34msource�[0m - 2021-11-04 22:47:23 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:23 �[32mINFO�[m i.a.i.s.j.AbstractJdbcSource(lambda$getCheckOperations$1):87 - {} - Attempting to get metadata from the database to see if we can connect.
�[34msource�[0m - 2021-11-04 22:47:23 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:23 �[32mINFO�[m i.a.i.s.r.CdcStateManager(<init>):26 - {} - Initialized CDC state with: null
�[34msource�[0m - 2021-11-04 22:47:23 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:23 �[32mINFO�[m i.a.i.s.r.StateManager(createCursorInfoForStream):118 - {} - No cursor field set in catalog but not present in state. Stream: AirbyteStreamNameNamespacePair{name='_airbyte_raw_asasdfasdfsdasapokemon', namespace='public'}, New Cursor Field: null. Resetting cursor value
�[34msource�[0m - 2021-11-04 22:47:23 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:23 �[32mINFO�[m i.a.i.s.r.StateManager(createCursorInfoForStream):118 - {} - No cursor field set in catalog but not present in state. Stream: AirbyteStreamNameNamespacePair{name='_airbyte_raw_asdfads_airbyte_raw_pokemon', namespace='public'}, New Cursor Field: null. Resetting cursor value
�[34msource�[0m - 2021-11-04 22:47:28 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:28 �[32mINFO�[m i.a.i.s.p.PostgresSource(isCdc):212 - {} - using CDC: false
�[34msource�[0m - 2021-11-04 22:47:28 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:28 �[32mINFO�[m i.a.i.s.r.AbstractRelationalDbSource(queryTableFullRefresh):35 - {} - Queueing query for table: _airbyte_raw_asasdfasdfsdasapokemon
�[34msource�[0m - 2021-11-04 22:47:28 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:28 �[32mINFO�[m i.a.i.s.r.AbstractRelationalDbSource(queryTableFullRefresh):35 - {} - Queueing query for table: _airbyte_raw_asdfads_airbyte_raw_pokemon
�[34msource�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.s.r.AbstractDbSource(lambda$read$2):121 - {} - Closing database connection pool.
�[34msource�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.s.r.AbstractDbSource(lambda$read$2):123 - {} - Closed database connection pool.
�[34msource�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):133 - {} - Completed integration: io.airbyte.integrations.base.ssh.SshWrappedSource
�[34msource�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.s.p.PostgresSource(main):264 - {} - completed source: class io.airbyte.integrations.source.postgres.PostgresSource
2021-11-04 22:47:31 INFO () DefaultReplicationWorker(run):121 - Source thread complete.
2021-11-04 22:47:31 INFO () DefaultReplicationWorker(run):122 - Waiting for destination thread to join.
�[35mdestination�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.b.FailureTrackingAirbyteMessageConsumer(close):80 - {} - Airbyte message consumer: succeeded.
�[35mdestination�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.d.l.LocalJsonDestination$JsonConsumer(close):192 - {} - finalizing consumer.
�[35mdestination�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.d.l.LocalJsonDestination$JsonConsumer(close):208 - {} - File output: /local/afdsds/_airbyte_raw__airbyte_raw_asdfads_airbyte_raw_pokemon.jsonl
�[35mdestination�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.d.l.LocalJsonDestination$JsonConsumer(close):208 - {} - File output: /local/afdsds/_airbyte_raw__airbyte_raw_asasdfasdfsdasapokemon.jsonl
�[35mdestination�[0m - 2021-11-04 22:47:31 INFO () DefaultAirbyteStreamFactory(lambda$create$0):61 - 2021-11-04 22:47:31 �[32mINFO�[m i.a.i.b.IntegrationRunner(run):153 - {} - Completed integration: io.airbyte.integrations.destination.local_json.LocalJsonDestination
2021-11-04 22:47:31 INFO () DefaultReplicationWorker(run):124 - Destination thread complete.
2021-11-04 22:47:31 INFO () DefaultReplicationWorker(run):152 - sync summary: io.airbyte.config.ReplicationAttemptSummary@2c59d72a[status=completed,recordsSynced=4,bytesSynced=100944,startTime=1636066036442,endTime=1636066051420]
2021-11-04 22:47:31 INFO () DefaultReplicationWorker(run):161 - Source did not output any state messages
2021-11-04 22:47:31 WARN () DefaultReplicationWorker(run):172 - State capture: No state retained.
2021-11-04 22:47:31 INFO () TemporalAttemptExecution(get):137 - Stopping cancellation check scheduling...
2021-11-04 22:47:31 INFO () SyncWorkflow$ReplicationActivityImpl(replicate):233 - sync summary: io.airbyte.config.StandardSyncOutput@7a9b0b84[standardSyncSummary=io.airbyte.config.StandardSyncSummary@61d1981e[status=completed,recordsSynced=4,bytesSynced=100944,startTime=1636066036442,endTime=1636066051420],state=<null>,outputCatalog=io.airbyte.protocol.models.ConfiguredAirbyteCatalog@30785316[streams=[io.airbyte.protocol.models.ConfiguredAirbyteStream@6c8e9a9b[stream=io.airbyte.protocol.models.AirbyteStream@4eeb6ba5[name=_airbyte_raw_asasdfasdfsdasapokemon,jsonSchema={"type":"object","properties":{"_airbyte_data":{"type":"string"},"_airbyte_ab_id":{"type":"string"},"_airbyte_emitted_at":{"type":"string"}}},supportedSyncModes=[full_refresh, incremental],sourceDefinedCursor=<null>,defaultCursorField=[],sourceDefinedPrimaryKey=[[_airbyte_ab_id]],namespace=public,additionalProperties={}],syncMode=full_refresh,cursorField=[],destinationSyncMode=append,primaryKey=[[_airbyte_ab_id]],additionalProperties={}], io.airbyte.protocol.models.ConfiguredAirbyteStream@6f7fa0d6[stream=io.airbyte.protocol.models.AirbyteStream@30a16528[name=_airbyte_raw_asdfads_airbyte_raw_pokemon,jsonSchema={"type":"object","properties":{"_airbyte_data":{"type":"string"},"_airbyte_ab_id":{"type":"string"},"_airbyte_emitted_at":{"type":"string"}}},supportedSyncModes=[full_refresh, incremental],sourceDefinedCursor=<null>,defaultCursorField=[],sourceDefinedPrimaryKey=[[_airbyte_ab_id]],namespace=public,additionalProperties={}],syncMode=full_refresh,cursorField=[],destinationSyncMode=append,primaryKey=[[_airbyte_ab_id]],additionalProperties={}]],additionalProperties={}]]