You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/17 01:13:19 UTC
(pulsar) branch master updated: [fix][fn] Fix Go function incorrectly using nanoseconds for timestamp (#25869)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new efe1465d7eac [fix][fn] Fix Go function incorrectly using nanoseconds for timestamp (#25869)
efe1465d7eac is described below
commit efe1465d7eacc89c01dd595888dee340423433b6
Author: Dream95 <86...@qq.com>
AuthorDate: Sat Oct 17 09:13:13 2026 +0800
[fix][fn] Fix Go function incorrectly using nanoseconds for timestamp (#25869)
Signed-off-by: Dream95 <zh...@163.com>
Signed-off-by: Matteo Merli <mm...@apache.org>
Co-authored-by: Matteo Merli <mm...@apache.org>
---
pulsar-function-go/pf/instance.go | 5 +++--
pulsar-function-go/pf/stats.go | 6 +++---
pulsar-function-go/pf/stats_test.go | 25 +++++++++++++++++++++++++
3 files changed, 31 insertions(+), 5 deletions(-)
diff --git a/pulsar-function-go/pf/instance.go b/pulsar-function-go/pf/instance.go
index f2b5dfd50b80..581c6d371b55 100644
--- a/pulsar-function-go/pf/instance.go
+++ b/pulsar-function-go/pf/instance.go
@@ -725,11 +725,12 @@ func (gi *goInstance) getAvgProcessLatency() float32 {
return float32(*sum) / float32(*count)
}
-func (gi *goInstance) getLastInvocation() float32 {
+func (gi *goInstance) getLastInvocation() float64 {
metric := gi.getMatchingMetricFromRegistry(PulsarFunctionMetricsPrefix + LastInvocation)
// "pulsar_function_" + "last_invocation", GaugeVec.
+ // float64, as a float32 would round an epoch-millis timestamp to ~2 minutes.
val := metric.GetGauge().Value
- return float32(*val)
+ return *val
}
func (gi *goInstance) getTotalProcessedSuccessfully1min() float32 {
diff --git a/pulsar-function-go/pf/stats.go b/pulsar-function-go/pf/stats.go
index d25424b37047..a1b7035b3b2a 100644
--- a/pulsar-function-go/pf/stats.go
+++ b/pulsar-function-go/pf/stats.go
@@ -248,7 +248,7 @@ func getFirstMatch(
func (stat *StatWithLabelValues) setLastInvocation() {
now := time.Now()
- stat.statLastInvocation.Set(float64(now.UnixNano()))
+ stat.statLastInvocation.Set(float64(now.UnixMilli()))
}
func (stat *StatWithLabelValues) processTimeStart() {
@@ -272,7 +272,7 @@ func (stat *StatWithLabelValues) incrTotalUserExceptions(err error) {
func (stat *StatWithLabelValues) addUserException(err error) {
now := time.Now()
- ts := now.UnixNano()
+ ts := now.UnixMilli()
errorTS := LatestException{err, ts}
stat.latestUserException = append(stat.latestUserException, errorTS)
if len(stat.latestUserException) > 10 {
@@ -302,7 +302,7 @@ func (stat *StatWithLabelValues) incrTotalSysExceptions(exception error) {
func (stat *StatWithLabelValues) addSysException(exception error) {
now := time.Now()
- ts := now.UnixNano()
+ ts := now.UnixMilli()
errorTS := LatestException{exception, ts}
stat.latestSysException = append(stat.latestSysException, errorTS)
if len(stat.latestSysException) > 10 {
diff --git a/pulsar-function-go/pf/stats_test.go b/pulsar-function-go/pf/stats_test.go
index 550ef18aa8f0..2c0f5b61778d 100644
--- a/pulsar-function-go/pf/stats_test.go
+++ b/pulsar-function-go/pf/stats_test.go
@@ -21,6 +21,7 @@ package pf
import (
"context"
+ "errors"
"fmt"
"io"
"math"
@@ -216,3 +217,27 @@ func TestInstanceControlMetrics(t *testing.T) {
assert.EqualValuesf(t, value+1, metrics.UserMetrics[label], "user metric %s != %d", label, value+1)
}
}
+
+func TestStatsTimestampsAreMillis(t *testing.T) {
+ gi := newGoInstance()
+ t.Cleanup(gi.close)
+
+ before := time.Now().UnixMilli()
+ gi.stats.setLastInvocation()
+ gi.stats.incrTotalUserExceptions(errors.New("user error"))
+ gi.stats.incrTotalSysExceptions(errors.New("system error"))
+ after := time.Now().UnixMilli()
+
+ status := gi.getFunctionStatus()
+ assert.GreaterOrEqual(t, status.LastInvocationTime, before)
+ assert.LessOrEqual(t, status.LastInvocationTime, after)
+ assert.Len(t, status.LatestUserExceptions, 1)
+ assert.GreaterOrEqual(t, status.LatestUserExceptions[0].MsSinceEpoch, before)
+ assert.LessOrEqual(t, status.LatestUserExceptions[0].MsSinceEpoch, after)
+ assert.Len(t, status.LatestSystemExceptions, 1)
+ assert.GreaterOrEqual(t, status.LatestSystemExceptions[0].MsSinceEpoch, before)
+ assert.LessOrEqual(t, status.LatestSystemExceptions[0].MsSinceEpoch, after)
+
+ metrics := gi.getMetrics()
+ assert.Equal(t, status.LastInvocationTime, metrics.LastInvocation)
+}