Add instrumentation and logging to GCS sink.

[lavkesh]

Acceptance criteria:

1. Analysis of metrics for GCS sink.
2. Implementation.
3. Tracing.

Metrics:

1. record_write_count , tags : filename(partition+uuid)
2. file_open_total
3. file_closed_total, tags: success(true/false)
4. file_closing_time_milliseconds
5. file_size_bytes_total
6. file_upload_total , tags: success(true/false)
7. file_upload_time_milliseconds
8. file_upload_bytes

Discussion: 
1. Distribution of the file size.
2. upload time.
3. success/failures of upload.
4. how many open files are there
5. time taken to close parquet files.
6. messages read/messages per parquet file.
7. No of Error messages or dlqed. 
8. Think about completeness/freshness/deduplication.

[fzrvic]

added metrics:

1. record_processed_count

   • tags :
     • stage(file_write/upload)
     • filename(partition+uuid)
     • partition
   • counter of how many records/message written to files (parquet) tags, can be used to inform how many records written and uploaded for each file

2. file_open_total

   • tags :
     • partition
   • counter of how many file is created (opened)

3. file_closed_total

   • tags:
     • partition
     • success(true/false)
   • counter of how many file closed, tags can be used to tell how many success/fails, is there any issues when closing a file

4. file_closing_time_milliseconds

   • tags :
     • filename(partition+uuid)
   • timer to record how long the duration of closing local file

5. file_size_bytes_total

   • tags:
     • filename(partition+uuid)
     • partition
   • counter of how many bytes file written and successfully closed, tags can be used to inform how many bytes written for each files/ partition

6. file_upload_total

   • tags:
     • success(true/false)
     • partition
   • counter of how many file is uploaded , tags will tell how many upload success/ fails, this will help us identify issues on object storage config/ other issues on object storage itself

7. file_upload_time_milliseconds

   • tags:
     • filename(partition+uuid)
   • duration of file upload

8. file_upload_bytes

   • tags:
     • filename(partition+uuid)
     • partition
   • size of files that uploaded

[gauravsinghania]

For file_upload_total, do we want to add error type as tag?

[sravankorumilli]

filename as tag shouldn't be there, as that won't be useful, imo we shouldn't choose something as a tag which is unique as time proceeds

[lavkesh]

Remove file-name from tag.

• As a rule tags should not be unbounded

validate topics in the path.
validate key for the writermap.
metric for local file storage failure.
add topic as tags.
error type as tags.