add sink and source prometheus stats

[jerrypeng]

Motivation

Currently, source and sink don't have their dedicated stats. The prometheus stats exposed for sources and sinks will still have the same stats and format of a function which does not make sense in many instances.

Modifications

In this PR, I Add dedicated prometheus stats for sources and sinks.

The changes i made:

1. All prometheus source stats will be prefixed with "pulsar_source_"
2. All prometheus sink stats will be prefixed with "pulsar_sink_"
3. The prometheus Functions stats also reported source and sink exceptions which is not appropriate so I removed them and add those exceptions to system stats
4. Add unit tests for sources and sinks
5. Fixed a folder name type for data-generator source and sink since they are needed for the unit test

Result

Sample of Source stats:

# HELP pulsar_source_written_total_1min Total number of records written to a Pulsar topic in the last 1 minute.
# TYPE pulsar_source_written_total_1min counter
pulsar_source_written_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 11.0
# HELP pulsar_source_system_exception Exception from system code.
# TYPE pulsar_source_system_exception gauge
# HELP pulsar_source_source_exception Exception from source.
# TYPE pulsar_source_source_exception gauge
# HELP pulsar_source_user_metric_ User defined metric.
# TYPE pulsar_source_user_metric_ summary
# HELP pulsar_source_last_invocation The timestamp of the last invocation of the source.
# TYPE pulsar_source_last_invocation gauge
pulsar_source_last_invocation{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 1.545986927622E12
# HELP pulsar_source_source_exceptions_total Total number of source exceptions.
# TYPE pulsar_source_source_exceptions_total counter
pulsar_source_source_exceptions_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 0.0
# HELP pulsar_source_system_exceptions_total_1min Total number of system exceptions in the last 1 minute.
# TYPE pulsar_source_system_exceptions_total_1min counter
pulsar_source_system_exceptions_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 0.0
# HELP pulsar_source_received_total Total number of records received from source.
# TYPE pulsar_source_received_total counter
pulsar_source_received_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 12.0
# HELP pulsar_source_written_total Total number of records written to a Pulsar topic.
# TYPE pulsar_source_written_total counter
pulsar_source_written_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 11.0
# HELP pulsar_source_source_exceptions_total_1min Total number of source exceptions in the last 1 minute.
# TYPE pulsar_source_source_exceptions_total_1min counter
pulsar_source_source_exceptions_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 0.0
# HELP pulsar_source_system_exceptions_total Total number of system exceptions.
# TYPE pulsar_source_system_exceptions_total counter
pulsar_source_system_exceptions_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 0.0
# HELP pulsar_source_received_total_1min Total number of records received from source in the last 1 minute.
# TYPE pulsar_source_received_total_1min counter
pulsar_source_received_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSource-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSource-test",} 12.0

Sample of Sink stats:

# HELP pulsar_sink_received_total_1min Total number of messages sink has received from Pulsar topic(s) in the last 1 minute.
# TYPE pulsar_sink_received_total_1min counter
pulsar_sink_received_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 10.0
# HELP pulsar_sink_sink_exceptions_total Total number of sink exceptions.
# TYPE pulsar_sink_sink_exceptions_total counter
pulsar_sink_sink_exceptions_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 0.0
# HELP pulsar_sink_received_total Total number of records sink has received from Pulsar topic(s).
# TYPE pulsar_sink_received_total counter
pulsar_sink_received_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 10.0
# HELP pulsar_sink_user_metric_ User defined metric.
# TYPE pulsar_sink_user_metric_ summary
# HELP pulsar_sink_system_exception Exception from system code.
# TYPE pulsar_sink_system_exception gauge
# HELP pulsar_sink_system_exceptions_total_1min Total number of system exceptions in the last 1 minute.
# TYPE pulsar_sink_system_exceptions_total_1min counter
pulsar_sink_system_exceptions_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 0.0
# HELP pulsar_sink_written_total Total number of records processed by sink.
# TYPE pulsar_sink_written_total counter
pulsar_sink_written_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 10.0
# HELP pulsar_sink_sink_exceptions_total_1min Total number of sink exceptions in the last 1 minute.
# TYPE pulsar_sink_sink_exceptions_total_1min counter
pulsar_sink_sink_exceptions_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 0.0
# HELP pulsar_sink_written_total_1min Total number of records processed by sink the last 1 minute.
# TYPE pulsar_sink_written_total_1min counter
pulsar_sink_written_total_1min{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 10.0
# HELP pulsar_sink_last_invocation The timestamp of the last invocation of the sink.
# TYPE pulsar_sink_last_invocation gauge
pulsar_sink_last_invocation{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 1.545987016932E12
# HELP pulsar_sink_sink_exception Exception from sink.
# TYPE pulsar_sink_sink_exception gauge
# HELP pulsar_sink_system_exceptions_total Total number of system exceptions.
# TYPE pulsar_sink_system_exceptions_total counter
pulsar_sink_system_exceptions_total{tenant="external-repl-prop",namespace="external-repl-prop/io",function="PulsarSink-test",instance_id="0",cluster="use",fqfn="external-repl-prop/io/PulsarSink-test",} 0.0

[jerrypeng]

rerun integration tests

[jerrypeng]

rerun integration tests