---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/cpp/examples/log_parser.md
  - https://ytsaurus.tech/docs/ru/flow/cpp/examples/log_parser.md
  - href: https://ytsaurus.tech/docs/en/flow/cpp/examples/log_parser.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: en/_includes/flow/cpp/examples/log_parser.md -->
# Log parser in YTsaurus Flow (C++)

This example shows a [process function](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md) in materialized ordered-source mode. The [pipeline](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/log_parser) reads log lines from a queue and parses them while reading the source, without an intermediate passthrough computation. The built-in `TProcessFunctionTransformOrderedSourceComputation` executes the function. The example also shows reading from a `Source` and maintaining durable state that survives restarts (see [State](#state)).

[Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/log_parser)

## Pipeline components

### TLogParserProcessFunction

The user logic is written as a [process function](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md): a subclass of `IProcessFunction` that doesn’t depend on the `Computation` object, so you can cover it with unit tests without a cluster ([unittest](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/log_parser/unittest/log_parser_process_function_ut.cpp)). It’s executed by the built-in `TProcessFunctionTransformOrderedSourceComputation` adapter, which also sets the mode to ordered source (see the [list of adapters](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md#how-it-works)).

In `ProcessMessage(const TInputMessageConstPtr& message, const IOutputCollectorPtr& output, const IRuntimeContextPtr& context)`, the function reads the `line` column of the raw `source` message via `GetColumnValue<std::string>(message, "line")` and parses it with `ParseLogLine` into records of the form `"level:text"` separated by `;`, discarding records without the `:` separator, with empty text, or with a level other than `info`, `warning`, and `error`. For each valid record, it updates the state through the `StateClient_.GetState(message->Key)` accessor (see [State](#state)), builds a `TLogRecordMessage`, and emits it into the `records` output stream by calling `output->AddMessage(context->ConvertToMessage(outputRecord))`.

The transformation result, the `records` stream, is materialized in YTsaurus as in `TTransformComputation`, so the transformation has no determinism requirements: after a restart, Flow distributes the already materialized messages with the `MessageId` values previously assigned to them instead of recomputing them.

The [Proto parser](https://ytsaurus.tech/docs/en/flow/cpp/examples/proto_parser.md) example uses the same hosting mode through a reusable process-function base: `TProtoLogParserFunction` inherits from `TProtoParsingProcessFunctionBase<TLogRecordProto>` and runs under `TProcessFunctionTransformOrderedSourceComputation`. The host adapter’s spec validator is the same as the computation base class’s: a non-empty `group_by_schema`, timers, key-visitor streams, and `external_state_managers` are rejected (for the full [list of limitations](https://ytsaurus.tech/docs/en/flow/cpp/computation.md#ttransformorderedsourcecomputation)).

### The parser computation spec

Two fields of the `parser` computation spec link the function to the adapter (see [Registration](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md#registration)):

```yson
"computation_class_name" = "NYT::NFlow::TProcessFunctionTransformOrderedSourceComputation";
"processing_function" = "NYT::NFlow::NExample::TLogParserProcessFunction";
```

The remaining fields of the `parser` record describe the connections: `source_streams.queue` is a `TQueueSource` with the paths to the queue and the consumer, and `sinks.queue` is a direct external `TSyncQueueSink` for the `records` stream, so no separate sink computation is needed. For the full file, see [pipeline.yson](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/log_parser/pipeline.yson).

## Message types

`TLogRecordMessage` is a subclass of `TYsonMessage` (a YSON structure registered via `YT_FLOW_DEFINE_YSON_MESSAGE`) with the following fields:

- `level`: the record level (`info`, `warning`, or `error`);
- `text`: the record text;
- `worst_level_so_far`: the highest severity level (`info < warning < error`) encountered in this source partition at the time of the record (see [State](#state)).

## State {#state}

`TLogParserProcessFunction` is stateful. It keeps the `TWorstSeverityState` state in the `TMutableStateKeyClient<TWorstSeverityState> StateClient_` field (see [Working with states (C++)](https://ytsaurus.tech/docs/en/flow/cpp/state.md#internal-state)). The `TProcessFunctionTransformOrderedSourceComputation` adapter calls `Init(const IRuntimeInitContextPtr& initContext)`, where the client connects to the state via `initContext->InitClient(StateClient_, WorstSeverityStateName)` (the state name is `worst_severity`), and `ProcessMessage`, where the state is read through `GetState(message->Key)` and output records are converted into messages via `context->ConvertToMessage(...)`.

The computation instance is bound to a single source partition, so all messages carry the same key and address the same state row: `state->WorstSeverity = std::max(state->WorstSeverity, SeverityRank(record.Level))`.

The framework synchronizes this state in the same epoch transaction as the advance of the source offset (see [Computation](https://ytsaurus.tech/docs/en/flow/cpp/computation.md#ttransformorderedsourcecomputation)); the function itself does nothing for this. Therefore, user state is exactly-once correct and doesn’t have to be idempotent under reprocessing: an ordinary, “naive” counter of processed records would be just as correct here. The example keeps a running maximum severity simply because it’s a natural aggregate for such a pipeline, not because it’s in any way safer than a counter.

## The main function

The `main` function does the following:
1. `NYT::NFlow::Initialize(argc, argv)`: initializes the Flow library.
2. `TSimpleSpecBuilder`: the builder for registering streams. `RegisterStream<TLogRecordMessage>("records")` registers the `records` stream with the `TLogRecordMessage` message type.
3. `TSimpleRunnerProgram(std::move(builder)).Run(argc, argv)`: starts the pipeline.

You don’t need to register the function in `main`: the `YT_FLOW_DEFINE_PROCESS_FUNCTION(TLogParserProcessFunction)` macro is placed at file level in `lib/log_parser_process_function.cpp`, and the file itself is linked into the library as `GLOBAL`, so the registry entry appears when the binary is initialized.

## Source code

### TLogParserProcessFunction

```cpp
#pragma once

#include "log_line_parser.h"

#include <yt/yt/flow/library/cpp/common/process_function.h>
#include <yt/yt/flow/library/cpp/common/runtime_init_context.h>
#include <yt/yt/flow/library/cpp/common/state_client.h>
#include <yt/yt/flow/library/cpp/common/yson_message.h>

#include <string>

namespace NYT::NFlow::NExample {

////////////////////////////////////////////////////////////////////////////////

struct TLogRecordMessage
    : public TYsonMessage
{
    std::string Level;
    std::string Text;
    std::string WorstLevelSoFar;

    REGISTER_YSON_STRUCT(TLogRecordMessage);

    static void Register(TRegistrar registrar)
    {
        registrar.Parameter("level", &TThis::Level)
            .Default();
        registrar.Parameter("text", &TThis::Text)
            .Default();
        registrar.Parameter("worst_level_so_far", &TThis::WorstLevelSoFar)
            .Default();
    }
};

////////////////////////////////////////////////////////////////////////////////

struct TWorstSeverityState
    : public NYTree::TYsonStruct
{
    int WorstSeverity = -1;

    REGISTER_YSON_STRUCT(TWorstSeverityState);

    static void Register(TRegistrar registrar)
    {
        registrar.Parameter("worst_severity", &TThis::WorstSeverity)
            .Default(-1);
    }
};

////////////////////////////////////////////////////////////////////////////////

inline constexpr TStringBuf WorstSeverityStateName = "worst_severity";

class TLogParserProcessFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override;

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& output,
        const IRuntimeContextPtr& context) override;

private:
    TMutableStateKeyClient<TWorstSeverityState> StateClient_;
};

////////////////////////////////////////////////////////////////////////////////

} // namespace NYT::NFlow::NExample
```

```cpp
#include "log_parser_process_function.h"

#include <yt/yt/flow/library/cpp/common/input_context.h>
#include <yt/yt/flow/library/cpp/common/registry.h>
#include <yt/yt/flow/library/cpp/common/runtime_context.h>

#include <algorithm>

namespace NYT::NFlow::NExample {

////////////////////////////////////////////////////////////////////////////////

void TLogParserProcessFunction::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitClient(StateClient_, WorstSeverityStateName);
}

void TLogParserProcessFunction::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto state = StateClient_.GetState(message->Key);
    for (const auto& record : ParseLogLine(GetColumnValue<std::string>(message, "line"))) {
        state->WorstSeverity = std::max(state->WorstSeverity, SeverityRank(record.Level));

        auto outputRecord = New<TLogRecordMessage>();
        outputRecord->Level = record.Level;
        outputRecord->Text = record.Text;
        outputRecord->WorstLevelSoFar = SeverityName(state->WorstSeverity);
        output->AddMessage(context->ConvertToMessage(outputRecord));
    }
}

////////////////////////////////////////////////////////////////////////////////

YT_FLOW_DEFINE_YSON_MESSAGE(TLogRecordMessage);
YT_FLOW_DEFINE_PROCESS_FUNCTION(TLogParserProcessFunction);

////////////////////////////////////////////////////////////////////////////////

} // namespace NYT::NFlow::NExample
```

### ParseLogLine

```cpp
#include "log_line_parser.h"

#include <util/string/split.h>

namespace NYT::NFlow::NExample {

////////////////////////////////////////////////////////////////////////////////

int SeverityRank(TStringBuf level)
{
    if (level == "info") {
        return 0;
    }
    if (level == "warning") {
        return 1;
    }
    if (level == "error") {
        return 2;
    }
    return -1;
}

std::string SeverityName(int rank)
{
    switch (rank) {
        case 0:
            return "info";
        case 1:
            return "warning";
        case 2:
            return "error";
        default:
            return {};
    }
}

std::vector<TLogRecord> ParseLogLine(const std::string& line)
{
    std::vector<TLogRecord> records;
    for (const auto& entry : StringSplitter(line).Split(';').SkipEmpty()) {
        TStringBuf token = entry.Token();
        auto separator = token.find(':');
        if (separator == TStringBuf::npos) {
            continue;
        }
        auto level = token.substr(0, separator);
        auto text = token.substr(separator + 1);
        if (text.empty() || SeverityRank(level) < 0) {
            continue;
        }
        records.push_back({.Level = std::string(level), .Text = std::string(text)});
    }
    return records;
}

////////////////////////////////////////////////////////////////////////////////

} // namespace NYT::NFlow::NExample
```

## See also

- [Getting started (C++)](https://ytsaurus.tech/docs/en/flow/cpp/getting-started.md)
- [Process function (C++)](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md)
- [Computation (C++)](https://ytsaurus.tech/docs/en/flow/cpp/computation.md#ttransformorderedsourcecomputation)
- [Computation (concept)](https://ytsaurus.tech/docs/en/flow/concepts/computation.md#ttransformorderedsourcecomputation)
- [Working with states (C++)](https://ytsaurus.tech/docs/en/flow/cpp/state.md)
<!-- endsource: en/_includes/flow/cpp/examples/log_parser.md -->