pulsar-functions/proto/src/main/proto/InstanceCommunication.proto

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
syntax = "proto3";
package proto;

import "google/protobuf/empty.proto";

option java_package = "org.apache.pulsar.functions.proto";
option java_outer_classname = "InstanceCommunication";

message FunctionStatus {
    message ExceptionInformation {
        string exceptionString = 1;
        int64 msSinceEpoch = 2;
    }
    bool running = 1;
    string failureException = 2;
    int64 numRestarts = 3;
    // int64 numProcessed = 4;
    int64 numReceived = 17;
    int64 numSuccessfullyProcessed = 5;
    int64 numUserExceptions = 6;
    repeated ExceptionInformation latestUserExceptions = 7;
    int64 numSystemExceptions = 8;
    repeated ExceptionInformation latestSystemExceptions = 9;
    int64 numSourceExceptions = 18;
    repeated ExceptionInformation latestSourceExceptions = 19;
    int64 numSinkExceptions = 20;
    repeated ExceptionInformation latestSinkExceptions = 21;
    // map from topic name to number of deserialization exceptions
//    map<string, int64> deserializationExceptions = 10;
    // number of serialization exceptions on the output
//    int64 serializationExceptions = 11;
    // average latency
    double averageLatency = 12;

    // When was the last time the function was invoked.
    // expressed in ms since epoch
    int64 lastInvocationTime = 13;
    string instanceId = 14;
//    MetricsData metrics = 15 [deprecated=true];
    // owner of function-instance
    string workerId = 16;
}

// Deprecated
message FunctionStatusList {
    string error = 2;
    repeated FunctionStatus functionStatusList = 1;
}

message MetricsData {
//    message DataDigest {
//        double count = 1;
//        double sum = 2;
//        double max = 3;
//        double min = 4;
//    }
//    map<string, DataDigest> metrics = 1 [deprecated=true];

    // Total number of records function received from source
    int64 receivedTotal = 2;

    int64 receivedTotal_1min = 10;

    // Total number of records processed
    // No longer used because processedSuccessfullyTotal and userExceptionsTotal add to it
    // int64 processedTotal = 3;

    // int64 processedTotal_1min = 11;

    // Total number of records successfully processed by user function
    int64 processedSuccessfullyTotal = 4;

    int64 processedSuccessfullyTotal_1min = 12;

    // Total number of system exceptions thrown
    int64 systemExceptionsTotal = 5;

    int64 systemExceptionsTotal_1min = 13;

    // Total number of user exceptions thrown
    int64 userExceptionsTotal = 6;

    int64 userExceptionsTotal_1min = 14;

    // Average process latency for function
    double avgProcessLatency = 7;

    double avgProcessLatency_1min = 15;

    // Timestamp of when the function was last invoked
    int64 lastInvocation = 8;

    // User defined metrics
    map<string, double> userMetrics = 9;
}

message HealthCheckResult {
    bool success = 1;
}

service InstanceControl {
    rpc GetFunctionStatus(google.protobuf.Empty) returns (FunctionStatus) {}
    rpc GetAndResetMetrics(google.protobuf.Empty) returns (MetricsData) {}
    rpc ResetMetrics(google.protobuf.Empty) returns (google.protobuf.Empty) {}
    rpc GetMetrics(google.protobuf.Empty) returns (MetricsData) {}
    rpc HealthCheck(google.protobuf.Empty) returns (HealthCheckResult) {}
}

message Metrics {
    message InstanceMetrics {
        string name = 1;
        int32 instanceId = 2;
        MetricsData metricsData = 3;
    }
    repeated InstanceMetrics metrics = 1;
}